This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new eb797e1111 [core] Support object table in Jdbc and Hive catalog (#8624)
eb797e1111 is described below
commit eb797e11119be30b56d975c1c3618c707eb3e7c3
Author: eye-gu <[email protected]>
AuthorDate: Thu Jul 16 13:47:39 2026 +0800
[core] Support object table in Jdbc and Hive catalog (#8624)
---
.../org/apache/paimon/catalog/AbstractCatalog.java | 54 ++++++++++++++--
.../java/org/apache/paimon/jdbc/JdbcCatalog.java | 75 ++++++++++++++++++++--
.../org/apache/paimon/jdbc/JdbcCatalogTest.java | 66 +++++++++++++++++++
.../java/org/apache/paimon/hive/HiveCatalog.java | 38 +++++++++++
.../org/apache/paimon/hive/HiveCatalogTest.java | 65 +++++++++++++++++++
5 files changed, 288 insertions(+), 10 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
index edcdfa8eec..6c97bb9808 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
@@ -43,8 +43,11 @@ import org.apache.paimon.table.FormatTable;
import org.apache.paimon.table.Instant;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.TableSnapshot;
+import org.apache.paimon.table.object.ObjectTable;
import org.apache.paimon.table.sink.TableCommitImpl;
import org.apache.paimon.table.system.SystemTableLoader;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.SnapshotNotExistException;
import org.slf4j.Logger;
@@ -449,10 +452,8 @@ public abstract class AbstractCatalog implements Catalog {
createFormatTable(identifier, schema);
break;
case OBJECT_TABLE:
- throw new UnsupportedOperationException(
- String.format(
- "Catalog %s cannot support object tables.",
- this.getClass().getName()));
+ createObjectTable(identifier, schema);
+ break;
}
}
@@ -766,6 +767,38 @@ public abstract class AbstractCatalog implements Catalog {
this.getClass().getName() + " currently does not support
format table");
}
+ /**
+ * Create an {@link ObjectTable} identified by the given {@link
Identifier}.
+ *
+ * @param identifier Path of the table
+ * @param schema Schema of the table
+ */
+ public void createObjectTable(Identifier identifier, Schema schema) {
+ throw new UnsupportedOperationException(
+ this.getClass().getName() + " currently does not support
object table");
+ }
+
+ /**
+ * Build a {@link Schema} with the fixed fields of {@link ObjectTable} and
the options from the
+ * given schema. Object Table has a fixed schema and does not support
custom fields, so the
+ * user-provided fields are ignored.
+ *
+ * @param schema the user-provided schema containing options
(type=object-table, path, etc.)
+ * @return a new schema with ObjectTable's fixed fields and the
user-provided options
+ */
+ protected static Schema buildObjectTableSchema(Schema schema) {
+ RowType schemaRowType = ObjectTable.SCHEMA;
+ Schema.Builder builder = Schema.newBuilder();
+ for (DataField field : schemaRowType.getFields()) {
+ builder.column(field.name(), field.type(), field.description());
+ }
+ builder.options(schema.options());
+ if (schema.comment() != null) {
+ builder.comment(schema.comment());
+ }
+ return builder.build();
+ }
+
/**
* Get warehouse path for specified database. If a catalog would like to
provide individual path
* for each database, this method can be `Override` in that catalog.
@@ -813,7 +846,11 @@ public abstract class AbstractCatalog implements Catalog {
}
private void validateCustomTablePath(Map<String, String> options) {
- if (!allowCustomTablePath() &&
options.containsKey(CoreOptions.PATH.key())) {
+ TableType tableType = Options.fromMap(options).get(TYPE);
+ boolean isObjectTable = tableType == TableType.OBJECT_TABLE;
+ if (!isObjectTable
+ && !allowCustomTablePath()
+ && options.containsKey(CoreOptions.PATH.key())) {
throw new UnsupportedOperationException(
String.format(
"The current catalog %s does not support
specifying the table path when creating a table.",
@@ -870,7 +907,12 @@ public abstract class AbstractCatalog implements Catalog {
return s.copy(branchOptions.toMap());
});
}
- schema.ifPresent(s -> s.options().put(PATH.key(),
tablePath.toString()));
+ schema.ifPresent(s -> putPathOption(s, tablePath));
return schema;
}
+
+ /** Set the PATH option on the loaded schema. */
+ protected void putPathOption(TableSchema schema, Path tablePath) {
+ schema.options().put(PATH.key(), tablePath.toString());
+ }
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java
b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java
index e489297462..79687c9368 100644
--- a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java
@@ -420,10 +420,65 @@ public class JdbcCatalog extends AbstractCatalog {
}
break;
case OBJECT_TABLE:
- throw new UnsupportedOperationException(
+ try {
+ runWithLock(
+ identifier,
+ () -> {
+ if (!validateTableNotExists(identifier,
ignoreIfExists)) {
+ return null;
+ }
+ createObjectTable(identifier, schema);
+ return null;
+ });
+ } catch (TableAlreadyExistException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to create table " +
identifier.getFullName(), e);
+ }
+ break;
+ }
+ }
+
+ @Override
+ public void createObjectTable(Identifier identifier, Schema schema) {
+ try {
+ Path tablePath = getTableLocation(identifier);
+ schema.options().putIfAbsent(PATH.key(), tablePath.toString());
+ Schema objectSchema = buildObjectTableSchema(schema);
+ fileIO.mkdirs(tablePath);
+
+ // Write schema file via SchemaManager.commit()
+ SchemaManager schemaManager = getSchemaManager(identifier);
+ TableSchema tableSchema = TableSchema.create(0, objectSchema);
+ if (!schemaManager.commit(tableSchema)) {
+ throw new RuntimeException(
+ "Failed to commit schema for object table " +
identifier);
+ }
+
+ // Register table in JDBC catalog
+ if (!JdbcUtils.insertTable(
+ connections,
+ catalogKey,
+ identifier.getDatabaseName(),
+ identifier.getTableName())) {
+ fileIO.deleteDirectoryQuietly(tablePath);
+ throw new RuntimeException(
String.format(
- "Catalog %s cannot support object tables.",
- this.getClass().getName()));
+ "Failed to create table %s in catalog %s",
+ identifier.getFullName(), catalogKey));
+ }
+ if (syncTableProperties()) {
+ JdbcUtils.insertTableProperties(
+ connections,
+ catalogKey,
+ identifier.getDatabaseName(),
+ identifier.getTableName(),
+ collectTableProperties(tableSchema));
+ }
+ LOG.debug("Successfully created object table: {}", identifier);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to create table " +
identifier.getFullName(), e);
}
}
@@ -665,6 +720,16 @@ public class JdbcCatalog extends AbstractCatalog {
}
}
+ /**
+ * Use {@code putIfAbsent} because a JDBC catalog stores the schema file
at the warehouse
+ * default path while an object table's PATH option may point to a
different user-specified data
+ * directory. Overwriting would wrongly redirect the object table to the
schema directory.
+ */
+ @Override
+ protected void putPathOption(TableSchema schema, Path tablePath) {
+ schema.options().putIfAbsent(PATH.key(), tablePath.toString());
+ }
+
@Override
protected TableSchema loadTableSchema(Identifier identifier) throws
TableNotExistException {
assertMainBranch(identifier);
@@ -876,7 +941,9 @@ public class JdbcCatalog extends AbstractCatalog {
}
private void validateCustomTablePath(Map<String, String> tableOptions) {
- if (!allowCustomTablePath() && tableOptions.containsKey(PATH.key())) {
+ TableType tableType = Options.fromMap(tableOptions).get(TYPE);
+ boolean isObjectTable = tableType == TableType.OBJECT_TABLE;
+ if (!isObjectTable && !allowCustomTablePath() &&
tableOptions.containsKey(PATH.key())) {
throw new UnsupportedOperationException(
String.format(
"The current catalog %s does not support
specifying the table path when creating a table.",
diff --git
a/paimon-core/src/test/java/org/apache/paimon/jdbc/JdbcCatalogTest.java
b/paimon-core/src/test/java/org/apache/paimon/jdbc/JdbcCatalogTest.java
index 3e6209220a..7f985f938e 100644
--- a/paimon-core/src/test/java/org/apache/paimon/jdbc/JdbcCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/jdbc/JdbcCatalogTest.java
@@ -18,6 +18,8 @@
package org.apache.paimon.jdbc;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.TableType;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.CatalogTestBase;
@@ -28,6 +30,7 @@ import org.apache.paimon.options.Options;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.table.Table;
+import org.apache.paimon.table.object.ObjectTable;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.view.View;
@@ -103,6 +106,69 @@ public class JdbcCatalogTest extends CatalogTestBase {
@Test
public void testGetTable() throws Exception {}
+ @Test
+ public void testObjectTable() throws Exception {
+ String databaseName = "object_table_db";
+ String tableName = "object_table";
+ catalog.createDatabase(databaseName, false);
+ Identifier identifier = Identifier.create(databaseName, tableName);
+
+ // Create object table with only options (type=object-table)
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.TYPE.key(), TableType.OBJECT_TABLE.toString());
+ Schema schema = Schema.newBuilder().options(options).build();
+
+ catalog.createTable(identifier, schema, false);
+
+ // Verify table exists in JDBC catalog
+ assertThat(catalog.listTables(databaseName)).contains(tableName);
+
+ // Verify getTable returns ObjectTable instance
+ Table table = catalog.getTable(identifier);
+ assertThat(table).isInstanceOf(ObjectTable.class);
+
+ ObjectTable objectTable = (ObjectTable) table;
+ // Verify fixed schema fields
+ assertThat(objectTable.rowType().getFieldNames())
+ .containsExactly("path", "name", "length", "mtime", "atime",
"owner");
+ // Verify location is set (defaults to table path)
+ assertThat(objectTable.location()).isNotNull();
+ // Verify options contain type=object-table
+ assertThat(objectTable.options().get(CoreOptions.TYPE.key()))
+ .isEqualTo(TableType.OBJECT_TABLE.toString());
+
+ // Drop table and verify it's gone
+ catalog.dropTable(identifier, false);
+ assertThat(catalog.listTables(databaseName)).doesNotContain(tableName);
+ }
+
+ @Test
+ public void testObjectTableWithCustomPath() throws Exception {
+ String databaseName = "object_table_custom_db";
+ String tableName = "object_table_custom";
+ catalog.createDatabase(databaseName, false);
+ Identifier identifier = Identifier.create(databaseName, tableName);
+
+ // Create object table with custom path
+ String customPath =
+ new Path(warehouse, databaseName + ".db/" + tableName +
"_data").toString();
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.TYPE.key(), TableType.OBJECT_TABLE.toString());
+ options.put(CoreOptions.PATH.key(), customPath);
+ Schema schema = Schema.newBuilder().options(options).build();
+
+ catalog.createTable(identifier, schema, false);
+
+ // Verify getTable returns ObjectTable with the custom location
+ Table table = catalog.getTable(identifier);
+ assertThat(table).isInstanceOf(ObjectTable.class);
+ ObjectTable objectTable = (ObjectTable) table;
+ assertThat(objectTable.location()).isEqualTo(customPath);
+
+ // Clean up
+ catalog.dropTable(identifier, false);
+ }
+
@Test
public void testDropTableWhenTablePathMissing() throws Exception {
String databaseName = "test_db";
diff --git
a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
index d21b961098..926fc024f2 100644
---
a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
+++
b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
@@ -105,6 +105,7 @@ import static
org.apache.hadoop.hive.serde.serdeConstants.FIELD_DELIM;
import static org.apache.paimon.CoreOptions.DATA_FILE_PATH_DIRECTORY;
import static org.apache.paimon.CoreOptions.FILE_FORMAT;
import static org.apache.paimon.CoreOptions.PARTITION_EXPIRATION_TIME;
+import static org.apache.paimon.CoreOptions.PATH;
import static org.apache.paimon.CoreOptions.TYPE;
import static org.apache.paimon.TableType.FORMAT_TABLE;
import static org.apache.paimon.catalog.CatalogUtils.checkNotBranch;
@@ -1044,6 +1045,43 @@ public class HiveCatalog extends AbstractCatalog {
}
}
+ @Override
+ public void createObjectTable(Identifier identifier, Schema schema) {
+ Pair<Path, Boolean> pair = initialTableLocation(schema.options(),
identifier);
+ Path location = pair.getLeft();
+ boolean externalTable = pair.getRight();
+ schema.options().putIfAbsent(PATH.key(), location.toString());
+ Schema objectSchema = buildObjectTableSchema(schema);
+ TableSchema newSchema = TableSchema.create(0, objectSchema);
+
+ try {
+ // Create schema directory and write schema file via
SchemaManager.commit()
+ FileIO tableFileIO = fileIO(location);
+ tableFileIO.mkdirs(location);
+ boolean committed =
+ runWithLock(
+ identifier,
+ () -> schemaManager(identifier,
location).commit(newSchema));
+ if (!committed) {
+ throw new RuntimeException(
+ "Failed to commit schema for object table " +
identifier);
+ }
+
+ // Create HMS table
+ Table hiveTable = createHiveTable(identifier, newSchema, location,
externalTable);
+ clients().execute(client -> client.createTable(hiveTable));
+ } catch (Exception e) {
+ if (!externalTable) {
+ try {
+ fileIO(location).deleteDirectoryQuietly(location);
+ } catch (Exception ee) {
+ LOG.error("Delete directory[{}] fail for table {}",
location, identifier, ee);
+ }
+ }
+ throw new RuntimeException("Failed to create table " +
identifier.getFullName(), e);
+ }
+ }
+
private boolean usingExternalTable(Map<String, String> tableOptions) {
CatalogTableType tableType =
OptionsUtils.convertToEnum(
diff --git
a/paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/HiveCatalogTest.java
b/paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/HiveCatalogTest.java
index a6f7dd7d1c..e1053c4715 100644
---
a/paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/HiveCatalogTest.java
+++
b/paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/HiveCatalogTest.java
@@ -19,6 +19,7 @@
package org.apache.paimon.hive;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.TableType;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.CatalogTestBase;
@@ -31,6 +32,7 @@ import org.apache.paimon.partition.Partition;
import org.apache.paimon.partition.PartitionStatistics;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.table.object.ObjectTable;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.utils.CommonTestUtils;
@@ -456,6 +458,69 @@ public class HiveCatalogTest extends CatalogTestBase {
assertThat(actual.lastFileCreationTime()).isEqualTo(expected.lastFileCreationTime()
/ 1000);
}
+ @Test
+ public void testObjectTable() throws Exception {
+ String databaseName = "object_table_db";
+ String tableName = "object_table";
+ catalog.createDatabase(databaseName, false);
+ Identifier identifier = Identifier.create(databaseName, tableName);
+
+ // Create object table with only options (type=object-table)
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.TYPE.key(), TableType.OBJECT_TABLE.toString());
+ Schema schema = Schema.newBuilder().options(options).build();
+
+ catalog.createTable(identifier, schema, false);
+
+ // Verify table exists in HMS
+ assertThat(catalog.listTables(databaseName)).contains(tableName);
+
+ // Verify getTable returns ObjectTable instance
+ org.apache.paimon.table.Table table = catalog.getTable(identifier);
+ assertThat(table).isInstanceOf(ObjectTable.class);
+
+ ObjectTable objectTable = (ObjectTable) table;
+ // Verify fixed schema fields
+ assertThat(objectTable.rowType().getFieldNames())
+ .containsExactly("path", "name", "length", "mtime", "atime",
"owner");
+ // Verify location is set (defaults to table path)
+ assertThat(objectTable.location()).isNotNull();
+ // Verify options contain type=object-table
+ assertThat(objectTable.options().get(CoreOptions.TYPE.key()))
+ .isEqualTo(TableType.OBJECT_TABLE.toString());
+
+ // Drop table and verify it's gone
+ catalog.dropTable(identifier, false);
+ assertThat(catalog.listTables(databaseName)).doesNotContain(tableName);
+ }
+
+ @Test
+ public void testObjectTableWithCustomPath() throws Exception {
+ String databaseName = "object_table_custom_db";
+ String tableName = "object_table_custom";
+ catalog.createDatabase(databaseName, false);
+ Identifier identifier = Identifier.create(databaseName, tableName);
+
+ // Create object table with custom path
+ String customPath =
+ new Path(warehouse, databaseName + ".db/" + tableName +
"_data").toString();
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.TYPE.key(), TableType.OBJECT_TABLE.toString());
+ options.put(CoreOptions.PATH.key(), customPath);
+ Schema schema = Schema.newBuilder().options(options).build();
+
+ catalog.createTable(identifier, schema, false);
+
+ // Verify getTable returns ObjectTable with the custom location
+ org.apache.paimon.table.Table table = catalog.getTable(identifier);
+ assertThat(table).isInstanceOf(ObjectTable.class);
+ ObjectTable objectTable = (ObjectTable) table;
+ assertThat(objectTable.location()).isEqualTo(customPath);
+
+ // Clean up
+ catalog.dropTable(identifier, false);
+ }
+
@Test
public void testCreateExternalTableWithLocation(@TempDir
java.nio.file.Path tempDir)
throws Exception {
