hudi-agent commented on code in PR #20012:
URL: https://github.com/apache/hudi/pull/20012#discussion_r4078413965
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiTableProperties.java:
##########
@@ -55,6 +93,99 @@ public HudiTableProperties()
.map(name -> name.toLowerCase(ENGLISH))
.collect(toImmutableList()),
value -> value))
+ .add(enumProperty(
+ TABLE_TYPE_PROPERTY,
+ "Hudi table type",
+ HoodieTableType.class,
Review Comment:
🤖 nit: `partitioned_by`, `primary_key` and `ordering_fields` now repeat the
same 10-line lower-casing `ArrayType(VARCHAR)` PropertyMetadata block; it might
be worth a small `lowerCasedStringListProperty(name, description)` helper so
the three stay in sync.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiMetadata.java:
##########
@@ -304,6 +321,202 @@ public Optional<Object> getInfo(ConnectorSession session,
ConnectorTableHandle t
return Optional.of(new HudiTableInfo(table.getSchemaTableName(),
table.getTableType().name(), table.getBasePath()));
}
+ /**
+ * Creates an empty table: its {@code .hoodie} directory on storage and
its catalog entry.
+ * <p>
+ * This is the single-call path only. {@code beginCreateTable}/{@code
finishCreateTable} and a
+ * page sink are deliberately absent, so {@code CREATE TABLE AS} and
{@code INSERT} still fail as
+ * unsupported; no rows are written here.
+ * <p>
+ * An explicit {@code location} makes the table external, matching Hudi's
Spark SQL behaviour. An
+ * omitted one makes it managed, at {@code <schemaLocation>/<tableName>},
which is what decides
+ * whether {@code DROP TABLE} later deletes the data.
+ */
+ @Override
+ public void createTable(ConnectorSession session, ConnectorTableMetadata
tableMetadata, SaveMode saveMode)
+ {
+ SchemaTableName schemaTableName = tableMetadata.getTable();
+ if (saveMode == SaveMode.REPLACE) {
+ // Replacing a table means deciding what happens to the rows
already in it, which is the
+ // write path this connector does not yet have.
+ throw new TrinoException(NOT_SUPPORTED, "This connector does not
support replacing tables");
+ }
+ Database database =
metastore.getDatabase(schemaTableName.getSchemaName())
+ .orElseThrow(() -> new
SchemaNotFoundException(schemaTableName.getSchemaName()));
+ if (metastore.getTable(schemaTableName.getSchemaName(),
schemaTableName.getTableName()).isPresent()) {
+ if (saveMode == SaveMode.IGNORE) {
+ return;
+ }
+ throw new TrinoException(ALREADY_EXISTS, "Table already exists: "
+ schemaTableName);
+ }
+
+ // Everything that can be rejected is rejected before storage is
touched, so a bad statement
+ // leaves nothing behind.
+ HudiTableValidation.validateCreateTable(tableMetadata);
+
+ Map<String, Object> properties = tableMetadata.getProperties();
+ Optional<String> explicitLocation = getTableLocation(properties);
+ boolean external = explicitLocation.isPresent();
+ String basePath = explicitLocation.orElseGet(() ->
defaultTableLocation(database, schemaTableName));
+
+ TrinoFileSystem fileSystem = fileSystemFactory.create(session);
+ checkLocationIsEmpty(fileSystem, basePath);
+
+ // One schema object produces both hoodie.table.create.schema and the
metastore column list,
+ // so the two cannot disagree (HUDI-9435).
+ HoodieSchema tableSchema = HudiSchemaConverter.toTableSchema(
+ tableMetadata.getColumns(), schemaTableName.getTableName());
+ Table table = HudiMetastoreTables.buildTable(
+ schemaTableName,
+ basePath,
+ getTableType(properties),
+ tableSchema,
+ getPartitionedBy(properties),
+ external,
+ Optional.of(session.getUser()),
+ tableMetadata.getComment());
+ try {
+ HudiTableInitializer.initializeTable(fileSystem, basePath,
tableMetadata, tableSchema);
+ metastore.createTable(table, NO_PRIVILEGES);
+ }
+ catch (TableAlreadyExistsException e) {
+ // Another CREATE TABLE may have initialized the same managed
location and won the
+ // metastore race. Its catalog entry now owns .hoodie, so deleting
it would corrupt the
+ // live table. A different location still belongs to this failed
attempt and is safe to
+ // clean up.
+ if (!isTableRegisteredAtLocation(schemaTableName, basePath, e)) {
+ cleanupTableMetadata(fileSystem, basePath, e);
+ }
+ throw e;
+ }
+ catch (RuntimeException e) {
+ // Initialization may have written some or all of .hoodie, but no
catalog entry from
+ // this call references it. Left there it would make a retry fail
the emptiness check.
+ //
+ // Only .hoodie is removed, never the base path: the base path may
have been created by
+ // someone else, and this connector did not create it. On object
storage there is no
+ // directory to delete in any case -- deleteDirectory removes the
objects under the
+ // prefix, which is exactly the set initTable wrote or may have
partially written.
+ cleanupTableMetadata(fileSystem, basePath, e);
+ throw e;
+ }
+ }
+
+ private boolean isTableRegisteredAtLocation(SchemaTableName tableName,
String basePath, RuntimeException failure)
Review Comment:
🤖 nit: this returns `true` when the lookup fails, so the name reads as a
fact it can't actually establish. Could you rename to something like
`mayBeOwnedByAnotherTable` (or `isStorageOwnershipUncertainOrClaimed`) so the
call site's `!` reads as "safe to clean up"?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiMetadata.java:
##########
@@ -304,6 +321,202 @@ public Optional<Object> getInfo(ConnectorSession session,
ConnectorTableHandle t
return Optional.of(new HudiTableInfo(table.getSchemaTableName(),
table.getTableType().name(), table.getBasePath()));
}
+ /**
+ * Creates an empty table: its {@code .hoodie} directory on storage and
its catalog entry.
+ * <p>
+ * This is the single-call path only. {@code beginCreateTable}/{@code
finishCreateTable} and a
+ * page sink are deliberately absent, so {@code CREATE TABLE AS} and
{@code INSERT} still fail as
+ * unsupported; no rows are written here.
+ * <p>
+ * An explicit {@code location} makes the table external, matching Hudi's
Spark SQL behaviour. An
+ * omitted one makes it managed, at {@code <schemaLocation>/<tableName>},
which is what decides
+ * whether {@code DROP TABLE} later deletes the data.
+ */
+ @Override
+ public void createTable(ConnectorSession session, ConnectorTableMetadata
tableMetadata, SaveMode saveMode)
+ {
+ SchemaTableName schemaTableName = tableMetadata.getTable();
+ if (saveMode == SaveMode.REPLACE) {
+ // Replacing a table means deciding what happens to the rows
already in it, which is the
+ // write path this connector does not yet have.
+ throw new TrinoException(NOT_SUPPORTED, "This connector does not
support replacing tables");
+ }
+ Database database =
metastore.getDatabase(schemaTableName.getSchemaName())
+ .orElseThrow(() -> new
SchemaNotFoundException(schemaTableName.getSchemaName()));
+ if (metastore.getTable(schemaTableName.getSchemaName(),
schemaTableName.getTableName()).isPresent()) {
+ if (saveMode == SaveMode.IGNORE) {
+ return;
+ }
+ throw new TrinoException(ALREADY_EXISTS, "Table already exists: "
+ schemaTableName);
+ }
+
+ // Everything that can be rejected is rejected before storage is
touched, so a bad statement
+ // leaves nothing behind.
+ HudiTableValidation.validateCreateTable(tableMetadata);
+
+ Map<String, Object> properties = tableMetadata.getProperties();
+ Optional<String> explicitLocation = getTableLocation(properties);
+ boolean external = explicitLocation.isPresent();
+ String basePath = explicitLocation.orElseGet(() ->
defaultTableLocation(database, schemaTableName));
+
+ TrinoFileSystem fileSystem = fileSystemFactory.create(session);
+ checkLocationIsEmpty(fileSystem, basePath);
+
+ // One schema object produces both hoodie.table.create.schema and the
metastore column list,
+ // so the two cannot disagree (HUDI-9435).
+ HoodieSchema tableSchema = HudiSchemaConverter.toTableSchema(
+ tableMetadata.getColumns(), schemaTableName.getTableName());
+ Table table = HudiMetastoreTables.buildTable(
+ schemaTableName,
+ basePath,
+ getTableType(properties),
+ tableSchema,
+ getPartitionedBy(properties),
+ external,
+ Optional.of(session.getUser()),
+ tableMetadata.getComment());
+ try {
+ HudiTableInitializer.initializeTable(fileSystem, basePath,
tableMetadata, tableSchema);
+ metastore.createTable(table, NO_PRIVILEGES);
+ }
+ catch (TableAlreadyExistsException e) {
+ // Another CREATE TABLE may have initialized the same managed
location and won the
+ // metastore race. Its catalog entry now owns .hoodie, so deleting
it would corrupt the
+ // live table. A different location still belongs to this failed
attempt and is safe to
+ // clean up.
+ if (!isTableRegisteredAtLocation(schemaTableName, basePath, e)) {
+ cleanupTableMetadata(fileSystem, basePath, e);
+ }
+ throw e;
+ }
+ catch (RuntimeException e) {
+ // Initialization may have written some or all of .hoodie, but no
catalog entry from
+ // this call references it. Left there it would make a retry fail
the emptiness check.
+ //
+ // Only .hoodie is removed, never the base path: the base path may
have been created by
+ // someone else, and this connector did not create it. On object
storage there is no
+ // directory to delete in any case -- deleteDirectory removes the
objects under the
+ // prefix, which is exactly the set initTable wrote or may have
partially written.
+ cleanupTableMetadata(fileSystem, basePath, e);
+ throw e;
+ }
+ }
+
+ private boolean isTableRegisteredAtLocation(SchemaTableName tableName,
String basePath, RuntimeException failure)
+ {
+ try {
+ return metastore.getTable(tableName.getSchemaName(),
tableName.getTableName())
+ .map(table ->
table.getStorage().getLocation().equals(basePath))
Review Comment:
🤖 Could this string equality miss when the metastore normalizes the location
it stores? Thrift HMS rewrites `sd.location` through `Warehouse.getDnsPath` on
create (e.g. `file:///tmp/x` → `file:/tmp/x`, trailing slash stripped), so a
winner at the *same* path could compare unequal here and the loser would then
delete the winner's `.hoodie`. Comparing normalized paths (e.g. via
`Location.of(...)`/`StoragePath`) or erring towards "registered" on any
mismatch under the same table name might be safer, since the failure mode is
destructive.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiMetastoreTables.java:
##########
@@ -0,0 +1,133 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import io.trino.metastore.Column;
+import io.trino.metastore.StorageFormat;
+import io.trino.metastore.Table;
+import io.trino.spi.TrinoException;
+import io.trino.spi.connector.SchemaTableName;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaField;
+import org.apache.hudi.sync.common.util.HoodieMetastoreTableDescriptor;
+
+import java.util.List;
+import java.util.Optional;
+
+import static io.trino.plugin.hive.TableType.EXTERNAL_TABLE;
+import static io.trino.plugin.hive.TableType.MANAGED_TABLE;
+import static io.trino.plugin.hudi.HudiUtil.toColumnHandle;
+import static io.trino.spi.StandardErrorCode.INVALID_TABLE_PROPERTY;
+
+/**
+ * Translates {@link HoodieMetastoreTableDescriptor} into a Trino {@link
Table}.
+ * <p>
+ * The assembly decisions -- which input format carries the table type, which
columns are partition
+ * columns, which properties make the table recognisable to Spark SQL, where
+ * {@code serialization.format} goes -- are not made here. They live in {@code
hudi-sync-common}
+ * next to {@code SparkDataSourceTableUtils}, in Hudi and JDK types, so that
the normalization contract
+ * is reusable by metastore integrations without depending on Trino types. All
that happens here is
+ * the mapping into {@code io.trino.metastore} types. Parity tests in {@code
hudi-hive-sync} pin the
+ * shared format names and schema-splitting default to the values its existing
executor emits.
+ * <p>
+ * The columns come from the same {@link HoodieSchema} that becomes
+ * {@code hoodie.table.create.schema}, which is the one thing this class does
insist on. The
+ * connector keeps two independent descriptions of a table's columns -- that
schema, and the
+ * metastore storage descriptor -- and {@link HudiMetadata#getColumnHandles}
reads only the latter.
+ * Nothing reconciles them, so a table built from two separately-handled
inputs can end up registered
+ * with columns that do not match its own schema, or with none at all
(HUDI-9435:
+ * {@code SELECT * not allowed from relation that has no columns}). Both
{@code CREATE TABLE} and
+ * {@code register_table} go through here so that cannot happen.
+ * <p>
+ * Column types come from {@link HudiUtil#toColumnHandle}, the same
Avro-to-Trino mapping the read
+ * path uses, so the types registered are the types a query will see.
+ */
+public final class HudiMetastoreTables
+{
+ private HudiMetastoreTables() {}
+
+ /**
+ * Builds the metastore descriptor for a table's snapshot view.
+ * <p>
+ * Lenient about a partition column that is absent from {@code
tableSchema}, typing it as a
+ * string, because the shared layer is and because {@code register_table}
has to cope with real
+ * tables built by key generators that do not record their partition
fields in the schema.
+ * {@code CREATE TABLE} rejects that case earlier, in {@link
HudiTableValidation}, where the
+ * offending property can be named.
+ *
+ * @param tableSchema the table's Hudi schema, including the meta fields;
supplies both the
+ * column names and their types
+ * @param partitionedBy partition column names, in partition-path order
+ * @param external whether the table is external. An explicit {@code
location} makes a table
+ * external, an omitted one makes it managed; see {@link
HudiMetadata#createTable}.
+ */
+ public static Table buildTable(
+ SchemaTableName tableName,
+ String basePath,
+ HoodieTableType tableType,
+ HoodieSchema tableSchema,
+ List<String> partitionedBy,
+ boolean external,
+ Optional<String> owner,
+ Optional<String> comment)
+ {
+ HoodieMetastoreTableDescriptor descriptor;
+ try {
+ descriptor = HoodieMetastoreTableDescriptor.forSnapshotView(
+ tableSchema, partitionedBy, tableType, basePath, external);
+ }
+ catch (IllegalArgumentException e) {
+ throw new TrinoException(INVALID_TABLE_PROPERTY, e.getMessage(),
e);
+ }
+
+ ImmutableMap.Builder<String, String> parameters =
ImmutableMap.<String, String>builder()
+ .putAll(descriptor.getTableParameters());
+ comment.ifPresent(value -> parameters.put(Table.TABLE_COMMENT, value));
+
+ return Table.builder()
+ .setDatabaseName(tableName.getSchemaName())
+ .setTableName(tableName.getTableName())
+ // Hive treats the table type and the EXTERNAL parameter as
independent; the shared
+ // layer supplies the parameter, this supplies the type, and
they must agree.
+ .setTableType((external ? EXTERNAL_TABLE :
MANAGED_TABLE).name())
+ .setOwner(owner)
+ .setDataColumns(toColumns(descriptor.getDataFields()))
+
.setPartitionColumns(toColumns(descriptor.getPartitionFields()))
+ .setParameters(parameters.buildKeepingLast())
+ .withStorage(storage -> storage
+ .setStorageFormat(StorageFormat.create(
+ descriptor.getSerdeClassName(),
+ descriptor.getInputFormatClassName(),
+ descriptor.getOutputFormatClassName()))
+ .setSerdeParameters(descriptor.getSerdeParameters())
+ .setLocation(basePath))
+ .build();
+ }
+
+ private static List<Column> toColumns(List<HoodieSchemaField> fields)
+ {
+ ImmutableList.Builder<Column> columns =
ImmutableList.builderWithExpectedSize(fields.size());
+ for (HoodieSchemaField field : fields) {
+ columns.add(new Column(
+ field.name(),
+ toColumnHandle(field).getHiveType(),
+ field.doc().isPresent() ? Optional.of(field.doc().get()) :
Optional.empty(),
Review Comment:
🤖 nit: `field.doc().isPresent() ? Optional.of(field.doc().get()) :
Optional.empty()` could be `field.doc().toJavaOptional()` — Hudi's `Option`
already provides that conversion.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
