fileCollector)
+ throws IOException {
+ FileStatus[] files = fileIO.listStatus(directory);
+ if (files == null) {
+ return;
+ }
+
+ for (FileStatus file : files) {
+ if (file.isDir()) {
+ listAllFiles(fileIO, file.getPath(), fileCollector);
+ } else {
+ fileCollector.add(file);
+ }
+ }
+ }
+
+ private static InternalRow toRow(FileStatus file) {
+ return toRow(
+ file.getPath().toString(),
+ file.getPath().getName(),
+ file.getLen(),
+ Timestamp.fromEpochMillis(file.getModificationTime()),
+ Timestamp.fromEpochMillis(file.getAccessTime()),
+ file.getOwner(),
+ null,
+ null,
+ null,
+ null,
+ null,
+ new GenericMap(Collections.emptyMap()));
+ }
+
+ public static GenericRow toRow(Object... values) {
+ GenericRow row = new GenericRow(values.length);
+
+ for (int i = 0; i < values.length; ++i) {
+ Object value = values[i];
+ if (value instanceof String) {
+ value = BinaryString.fromString((String) value);
+ }
+ row.setField(i, value);
+ }
+
+ return row;
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/object/ObjectTable.java b/paimon-core/src/main/java/org/apache/paimon/table/object/ObjectTable.java
new file mode 100644
index 000000000000..65689108caae
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/object/ObjectTable.java
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.object;
+
+import org.apache.paimon.manifest.ManifestCacheFilter;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.DelegatedFileStoreTable;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.apache.paimon.table.sink.TableCommitImpl;
+import org.apache.paimon.table.sink.TableWriteImpl;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import java.util.HashSet;
+import java.util.Map;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/**
+ * A object table refers to a directory that contains multiple objects (files), Object table
+ * provides metadata indexes for unstructured data objects in this directory. Allowing users to
+ * analyze unstructured data in Object Storage.
+ *
+ * Object Table stores the metadata of objects in the underlying table.
+ */
+public interface ObjectTable extends FileStoreTable {
+
+ RowType SCHEMA =
+ RowType.builder()
+ .field("path", DataTypes.STRING().notNull())
+ .field("name", DataTypes.STRING().notNull())
+ .field("length", DataTypes.BIGINT().notNull())
+ .field("mtime", DataTypes.TIMESTAMP_LTZ_MILLIS())
+ .field("atime", DataTypes.TIMESTAMP_LTZ_MILLIS())
+ .field("owner", DataTypes.STRING().nullable())
+ .field("generation", DataTypes.INT().nullable())
+ .field("content_type", DataTypes.STRING().nullable())
+ .field("storage_class", DataTypes.STRING().nullable())
+ .field("md5_hash", DataTypes.STRING().nullable())
+ .field("metadata_mtime", DataTypes.TIMESTAMP_LTZ_MILLIS().nullable())
+ .field("metadata", DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()))
+ .build()
+ .notNull();
+
+ /** Object location in file system. */
+ String objectLocation();
+
+ /** Underlying table to store metadata. */
+ FileStoreTable underlyingTable();
+
+ long refresh();
+
+ @Override
+ ObjectTable copy(Map dynamicOptions);
+
+ /** Create a new builder for {@link ObjectTable}. */
+ static ObjectTable.Builder builder() {
+ return new ObjectTable.Builder();
+ }
+
+ /** Builder for {@link ObjectTable}. */
+ class Builder {
+
+ private FileStoreTable underlyingTable;
+ private String objectLocation;
+
+ public ObjectTable.Builder underlyingTable(FileStoreTable underlyingTable) {
+ this.underlyingTable = underlyingTable;
+ checkArgument(
+ new HashSet<>(SCHEMA.getFields())
+ .containsAll(underlyingTable.rowType().getFields()),
+ "Schema of Object Table should be %s, but is %s.",
+ SCHEMA,
+ underlyingTable.rowType());
+ return this;
+ }
+
+ public ObjectTable.Builder objectLocation(String objectLocation) {
+ this.objectLocation = objectLocation;
+ return this;
+ }
+
+ public ObjectTable build() {
+ return new ObjectTableImpl(underlyingTable, objectLocation);
+ }
+ }
+
+ /** An implementation for {@link ObjectTable}. */
+ class ObjectTableImpl extends DelegatedFileStoreTable implements ObjectTable {
+
+ private final String objectLocation;
+
+ public ObjectTableImpl(FileStoreTable underlyingTable, String objectLocation) {
+ super(underlyingTable);
+ this.objectLocation = objectLocation;
+ }
+
+ @Override
+ public BatchWriteBuilder newBatchWriteBuilder() {
+ throw new UnsupportedOperationException("Object table does not support Write.");
+ }
+
+ @Override
+ public StreamWriteBuilder newStreamWriteBuilder() {
+ throw new UnsupportedOperationException("Object table does not support Write.");
+ }
+
+ @Override
+ public TableWriteImpl newWrite(String commitUser) {
+ throw new UnsupportedOperationException("Object table does not support Write.");
+ }
+
+ @Override
+ public TableWriteImpl newWrite(String commitUser, ManifestCacheFilter manifestFilter) {
+ throw new UnsupportedOperationException("Object table does not support Write.");
+ }
+
+ @Override
+ public TableCommitImpl newCommit(String commitUser) {
+ throw new UnsupportedOperationException("Object table does not support Commit.");
+ }
+
+ @Override
+ public String objectLocation() {
+ return objectLocation;
+ }
+
+ @Override
+ public FileStoreTable underlyingTable() {
+ return wrapped;
+ }
+
+ @Override
+ public long refresh() {
+ try {
+ return ObjectRefresh.refresh(this);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public ObjectTable copy(Map dynamicOptions) {
+ return new ObjectTableImpl(wrapped.copy(dynamicOptions), objectLocation);
+ }
+
+ @Override
+ public FileStoreTable copy(TableSchema newTableSchema) {
+ return new ObjectTableImpl(wrapped.copy(newTableSchema), objectLocation);
+ }
+
+ @Override
+ public FileStoreTable copyWithoutTimeTravel(Map dynamicOptions) {
+ return new ObjectTableImpl(
+ wrapped.copyWithoutTimeTravel(dynamicOptions), objectLocation);
+ }
+
+ @Override
+ public FileStoreTable copyWithLatestSchema() {
+ return new ObjectTableImpl(wrapped.copyWithLatestSchema(), objectLocation);
+ }
+
+ @Override
+ public FileStoreTable switchToBranch(String branchName) {
+ return new ObjectTableImpl(wrapped.switchToBranch(branchName), objectLocation);
+ }
+ }
+}
diff --git a/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/HadoopCompliantFileIO.java b/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/HadoopCompliantFileIO.java
index 4d86c12a6e52..67027eabadfb 100644
--- a/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/HadoopCompliantFileIO.java
+++ b/paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/HadoopCompliantFileIO.java
@@ -286,5 +286,15 @@ public Path getPath() {
public long getModificationTime() {
return status.getModificationTime();
}
+
+ @Override
+ public long getAccessTime() {
+ return status.getAccessTime();
+ }
+
+ @Override
+ public String getOwner() {
+ return status.getOwner();
+ }
}
}
diff --git a/paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/HadoopCompliantFileIO.java b/paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/HadoopCompliantFileIO.java
index abfe0fabba61..80f3df582096 100644
--- a/paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/HadoopCompliantFileIO.java
+++ b/paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3/HadoopCompliantFileIO.java
@@ -286,5 +286,15 @@ public Path getPath() {
public long getModificationTime() {
return status.getModificationTime();
}
+
+ @Override
+ public long getAccessTime() {
+ return status.getAccessTime();
+ }
+
+ @Override
+ public String getOwner() {
+ return status.getOwner();
+ }
}
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkFileIO.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkFileIO.java
index db66379f3108..74512409bfc8 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkFileIO.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkFileIO.java
@@ -227,5 +227,10 @@ public Path getPath() {
public long getModificationTime() {
return status.getModificationTime();
}
+
+ @Override
+ public long getAccessTime() {
+ return status.getAccessTime();
+ }
}
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RefreshObjectTableProcedure.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RefreshObjectTableProcedure.java
new file mode 100644
index 000000000000..97eb3095f094
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RefreshObjectTableProcedure.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.procedure;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.table.object.ObjectTable;
+
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+import org.apache.flink.types.Row;
+
+/**
+ * Refresh Object Table procedure. Usage:
+ *
+ *
+ * CALL sys.refresh_object_table('tableId')
+ *
+ */
+public class RefreshObjectTableProcedure extends ProcedureBase {
+
+ private static final String IDENTIFIER = "refresh_object_table";
+
+ @ProcedureHint(argument = {@ArgumentHint(name = "table", type = @DataTypeHint("STRING"))})
+ public @DataTypeHint("ROW") Row[] call(
+ ProcedureContext procedureContext, String tableId)
+ throws Catalog.TableNotExistException {
+ ObjectTable table = (ObjectTable) table(tableId);
+ long fileNumber = table.refresh();
+ return new Row[] {Row.of(fileNumber)};
+ }
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory b/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
index 6fe5e74ebe0d..0ff3ac1f1e1c 100644
--- a/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
+++ b/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
@@ -79,3 +79,4 @@ org.apache.paimon.flink.procedure.FastForwardProcedure
org.apache.paimon.flink.procedure.MarkPartitionDoneProcedure
org.apache.paimon.flink.procedure.CloneProcedure
org.apache.paimon.flink.procedure.CompactManifestProcedure
+org.apache.paimon.flink.procedure.RefreshObjectTableProcedure
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ObjectTableITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ObjectTableITCase.java
new file mode 100644
index 000000000000..b9e30035b093
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ObjectTableITCase.java
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** ITCase for object table. */
+public class ObjectTableITCase extends CatalogITCaseBase {
+
+ @Test
+ public void testIllegalObjectTable() {
+ assertThatThrownBy(
+ () ->
+ sql(
+ "CREATE TABLE T (a INT, b INT, c INT) WITH ('type' = 'object-table')"))
+ .rootCause()
+ .hasMessageContaining("Schema of Object Table can be empty or");
+ assertThatThrownBy(() -> sql("CREATE TABLE T WITH ('type' = 'object-table')"))
+ .rootCause()
+ .hasMessageContaining("Object table should have object-location option.");
+ }
+
+ @Test
+ public void testObjectTableRefresh() throws IOException {
+ Path objectLocation = new Path(path + "/object-location");
+ FileIO fileIO = LocalFileIO.create();
+ sql(
+ "CREATE TABLE T WITH ('type' = 'object-table', 'object-location' = '%s')",
+ objectLocation);
+
+ // add new file
+ fileIO.overwriteFileUtf8(new Path(objectLocation, "f0"), "1,2,3");
+ sql("CALL sys.refresh_object_table('default.T')");
+ assertThat(sql("SELECT name, length FROM T")).containsExactlyInAnyOrder(Row.of("f0", 5L));
+
+ // add new file
+ fileIO.overwriteFileUtf8(new Path(objectLocation, "f1"), "4,5,6");
+ sql("CALL sys.refresh_object_table('default.T')");
+ assertThat(sql("SELECT name, length FROM T"))
+ .containsExactlyInAnyOrder(Row.of("f0", 5L), Row.of("f1", 5L));
+
+ // delete file
+ fileIO.deleteQuietly(new Path(objectLocation, "f0"));
+ sql("CALL sys.refresh_object_table('default.T')");
+ assertThat(sql("SELECT name, length FROM T")).containsExactlyInAnyOrder(Row.of("f1", 5L));
+
+ // time travel
+ assertThat(sql("SELECT name, length FROM T /*+ OPTIONS('scan.snapshot-id' = '1') */"))
+ .containsExactlyInAnyOrder(Row.of("f0", 5L));
+
+ // insert into
+ assertThatThrownBy(() -> sql("INSERT INTO T SELECT * FROM T"))
+ .rootCause()
+ .hasMessageContaining("Object table does not support Write.");
+ assertThat(sql("SELECT name, length FROM T")).containsExactlyInAnyOrder(Row.of("f1", 5L));
+ }
+}
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
index c93aad41a732..35b65a7b530b 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
@@ -35,6 +35,7 @@
import org.apache.paimon.spark.procedure.MigrateTableProcedure;
import org.apache.paimon.spark.procedure.Procedure;
import org.apache.paimon.spark.procedure.ProcedureBuilder;
+import org.apache.paimon.spark.procedure.RefreshObjectTableProcedure;
import org.apache.paimon.spark.procedure.RemoveOrphanFilesProcedure;
import org.apache.paimon.spark.procedure.RenameTagProcedure;
import org.apache.paimon.spark.procedure.RepairProcedure;
@@ -87,6 +88,7 @@ private static Map> initProcedureBuilders() {
procedureBuilders.put("reset_consumer", ResetConsumerProcedure::builder);
procedureBuilders.put("mark_partition_done", MarkPartitionDoneProcedure::builder);
procedureBuilders.put("compact_manifest", CompactManifestProcedure::builder);
+ procedureBuilders.put("refresh_object_table", RefreshObjectTableProcedure::builder);
return procedureBuilders.build();
}
}
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RefreshObjectTableProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RefreshObjectTableProcedure.java
new file mode 100644
index 000000000000..c6b6fdab4723
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RefreshObjectTableProcedure.java
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.procedure;
+
+import org.apache.paimon.table.object.ObjectTable;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+
+import static org.apache.spark.sql.types.DataTypes.StringType;
+
+/** Spark procedure to refresh Object Table. */
+public class RefreshObjectTableProcedure extends BaseProcedure {
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {ProcedureParameter.required("table", StringType)};
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("file_number", DataTypes.LongType, false, Metadata.empty())
+ });
+
+ protected RefreshObjectTableProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ Identifier tableIdent = toIdentifier(args.getString(0), PARAMETERS[0].name());
+ return modifyPaimonTable(
+ tableIdent,
+ table -> {
+ ObjectTable objectTable = (ObjectTable) table;
+ long fileNumber = objectTable.refresh();
+ InternalRow outputRow = newInternalRow(fileNumber);
+ return new InternalRow[] {outputRow};
+ });
+ }
+
+ public static ProcedureBuilder builder() {
+ return new Builder() {
+ @Override
+ public RefreshObjectTableProcedure doBuild() {
+ return new RefreshObjectTableProcedure(tableCatalog());
+ }
+ };
+ }
+
+ @Override
+ public String description() {
+ return "RefreshObjectTableProcedure";
+ }
+}
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/ObjectTableTest.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/ObjectTableTest.scala
new file mode 100644
index 000000000000..3a446e33d8c8
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/ObjectTableTest.scala
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.sql
+
+import org.apache.paimon.fs.Path
+import org.apache.paimon.fs.local.LocalFileIO
+import org.apache.paimon.spark.PaimonSparkTestBase
+
+import org.apache.spark.sql.Row
+
+class ObjectTableTest extends PaimonSparkTestBase {
+
+ test(s"Paimon object table") {
+ val objectLocation = new Path(tempDBDir + "/object-location")
+ val fileIO = LocalFileIO.create
+
+ spark.sql(s"""
+ |CREATE TABLE T TBLPROPERTIES (
+ | 'type' = 'object-table',
+ | 'object-location' = '$objectLocation'
+ |)
+ |""".stripMargin)
+
+ // add new file
+ fileIO.overwriteFileUtf8(new Path(objectLocation, "f0"), "1,2,3")
+ spark.sql("CALL sys.refresh_object_table('test.T')")
+ checkAnswer(
+ spark.sql("SELECT name, length FROM T"),
+ Row("f0", 5L) :: Nil
+ )
+
+ // add new file
+ fileIO.overwriteFileUtf8(new Path(objectLocation, "f1"), "4,5,6")
+ spark.sql("CALL sys.refresh_object_table('test.T')")
+ checkAnswer(
+ spark.sql("SELECT name, length FROM T"),
+ Row("f0", 5L) :: Row("f1", 5L) :: Nil
+ )
+
+ // time travel
+ checkAnswer(
+ spark.sql("SELECT name, length FROM T VERSION AS OF 1"),
+ Row("f0", 5L) :: Nil
+ )
+ }
+}