voonhous commented on code in PR #20012:
URL: https://github.com/apache/hudi/pull/20012#discussion_r4194837618
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiMetadata.java:
##########
@@ -304,6 +325,224 @@ 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 = Table.builder(HudiMetastoreTables.buildTable(
+ schemaTableName,
+ basePath,
+ getTableType(properties),
+ tableSchema,
+ getPartitionedBy(properties),
+ external,
+ Optional.of(session.getUser()),
+ tableMetadata.getComment()))
+ .setParameter(TRINO_QUERY_ID_NAME, session.getQueryId())
+ .setParameter(TRINO_MANAGED_TABLE_PARAMETER,
Boolean.toString(!external))
+ .build();
+ try {
+ HudiTableInitializer.initializeTable(fileSystem, basePath,
tableMetadata, tableSchema);
Review Comment:
**minor:** On S3, two same-name CREATEs can both pass
`checkLocationIsEmpty`, and the loser's `initTable` overwrites the winner's
`hoodie.properties`: `HoodieTableConfig.create` calls `storage.create(path)`
(overwrite=true) and `S3OutputFile.create()` is not exclusive. The loser then
correctly keeps storage, but the winner's catalog entry now sits on the loser's
table type, key generator and schema. HDFS/local fail here with
`FileAlreadyExistsException`.
Not blocking, but could `createTable` first claim the location with
`TrinoOutputFile.createExclusive` on a marker file, or document the S3 race in
the javadoc?
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiRegisterTableProcedure.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * 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 io.trino.filesystem.TrinoFileSystemFactory;
+import io.trino.metastore.Database;
+import io.trino.metastore.HiveMetastore;
+import io.trino.metastore.HiveMetastoreFactory;
+import io.trino.plugin.hudi.procedure.RegisterTableProcedure;
+import io.trino.plugin.hudi.procedure.UnregisterTableProcedure;
+import io.trino.spi.connector.ConnectorAccessControl;
+import io.trino.spi.connector.ConnectorSecurityContext;
+import io.trino.spi.connector.SchemaTableName;
+import io.trino.spi.security.AccessDeniedException;
+import io.trino.testing.AbstractTestQueryFramework;
+import io.trino.testing.QueryRunner;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.Proxy;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static io.trino.testing.TestingConnectorSession.SESSION;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+final class TestHudiRegisterTableProcedure
+ extends AbstractTestQueryFramework
Review Comment:
**minor:** This class starts a full `DistributedQueryRunner` for one SQL
test (`testDisabledByDefaultBeforeValidatingArguments`); the other tests call
the procedure directly, `TestHudiConfig` already pins the default, and
`TestHudiDdl` covers the `CALL` wiring.
Not blocking, but could that case invoke the procedure built from `new
HudiConfig()` and assert `PERMISSION_DENIED`, dropping
`AbstractTestQueryFramework`?
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiMetastoreTables.java:
##########
@@ -0,0 +1,154 @@
+/*
+ * 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.avro.Schema;
+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;
+import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED;
+
+/**
+ * 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) {
+ rejectLocalTimestamp(field.schema().toAvroSchema(), field.name());
+ columns.add(new Column(
+ field.name(),
+ toColumnHandle(field).getHiveType(),
+ field.doc().isPresent() ? Optional.of(field.doc().get()) :
Optional.empty(),
+ ImmutableMap.of()));
+ }
+ return columns.build();
+ }
+
+ private static void rejectLocalTimestamp(Schema schema, String columnPath)
+ {
+ if (schema.getLogicalType() != null &&
schema.getLogicalType().getName().startsWith("local-timestamp-")) {
+ // The Trino Avro mapping treats these as bigint, while hive-sync
writes TIMESTAMP.
+ // Refuse a metastore schema that would disagree with other Hudi
readers.
+ throw new TrinoException(NOT_SUPPORTED,
+ "Cannot map Hudi column " + columnPath + " with Avro
logical type " + schema.getLogicalType().getName()
+ + " to a consistent Hive type");
+ }
+ switch (schema.getType()) {
+ case RECORD -> schema.getFields().forEach(field ->
rejectLocalTimestamp(field.schema(), columnPath + "." + field.name()));
Review Comment:
**minor:** The RECORD/ARRAY/MAP/UNION recursion has no test;
`TestHudiDdl:157` only registers a top-level `local-timestamp` column, so
dropping the recursion would let a `struct<local-timestamp-micros>` register as
`bigint` unnoticed.
Not blocking, but could
`TestHudiMetastoreTables#testRejectsLogicalTypesWithoutHiveCounterpart` add a
nested-record field and an array-element case?
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMetadata.java:
##########
@@ -0,0 +1,412 @@
+/*
+ * 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 io.trino.filesystem.FileIterator;
+import io.trino.filesystem.Location;
+import io.trino.filesystem.TrinoFileSystem;
+import io.trino.filesystem.TrinoFileSystemFactory;
+import io.trino.filesystem.local.LocalFileSystemFactory;
+import io.trino.metastore.Database;
+import io.trino.metastore.HiveMetastore;
+import io.trino.metastore.Table;
+import io.trino.metastore.TableAlreadyExistsException;
+import io.trino.plugin.hudi.util.HudiSchemaConverter;
+import io.trino.spi.TrinoException;
+import io.trino.spi.connector.ColumnMetadata;
+import io.trino.spi.connector.ConnectorTableMetadata;
+import io.trino.spi.connector.SaveMode;
+import io.trino.spi.connector.SchemaTableName;
+import io.trino.spi.predicate.TupleDomain;
+import io.trino.spi.security.PrincipalType;
+import io.trino.spi.type.TypeManager;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Proxy;
+import java.nio.file.FileAlreadyExistsException;
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.OptionalLong;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.UnaryOperator;
+
+import static com.google.common.base.Throwables.getCausalChain;
+import static
com.google.common.util.concurrent.MoreExecutors.newDirectExecutorService;
+import static io.trino.plugin.hive.HiveMetadata.TRINO_QUERY_ID_NAME;
+import static io.trino.plugin.hive.TableType.EXTERNAL_TABLE;
+import static io.trino.plugin.hudi.HudiTableProperties.TABLE_TYPE_PROPERTY;
+import static io.trino.spi.type.BigintType.BIGINT;
+import static io.trino.testing.TestingConnectorSession.SESSION;
+import static org.apache.hudi.common.model.HoodieTableType.MERGE_ON_READ;
+import static org.apache.hudi.common.util.ConfigUtils.IS_QUERY_AS_RO_TABLE;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+final class TestHudiMetadata
+{
+ @TempDir
+ private Path temporaryDirectory;
+
+ @Test
+ void testCreateTableRaceDoesNotDeleteWinningTableMetadata()
+ {
+ CreateRace race =
createRace(TestHudiMetadata::asTableCreatedByAnotherQuery, false);
+
+ assertThatThrownBy(() -> race.metadata().createTable(SESSION,
race.tableMetadata(), SaveMode.FAIL))
+ .isInstanceOf(TableAlreadyExistsException.class);
+
+ assertThat(race.winningTable().get()).isNotNull();
+ assertThat(metadataExists(race, race.winningTable().get()))
+ .isTrue();
+ }
+
+ @Test
+ void
testCreateTableRetryRecognizesCommittedTableAfterLocationNormalization()
+ {
+ CreateRace race = createRace(table -> Table.builder(table)
+ .withStorage(storage -> storage.setLocation(
+ table.getStorage().getLocation().replace("local:///",
"local:/")))
+ .build(), false);
+
+ assertThatCode(() -> race.metadata().createTable(SESSION,
race.tableMetadata(), SaveMode.FAIL))
+ .doesNotThrowAnyException();
+
+ assertThat(race.attemptedTable().get()).isNotNull();
+ assertThat(race.attemptedTable().get().getParameters())
+ .containsEntry(TRINO_QUERY_ID_NAME, SESSION.getQueryId())
+ .containsEntry(HudiMetadata.TRINO_MANAGED_TABLE_PARAMETER,
"true");
+ assertThat(metadataExists(race, race.attemptedTable().get()))
+ .isTrue();
+ }
+
+ @Test
+ void testCreateTableRacePreservesMetadataWhenWinnerUsesDifferentLocation()
+ {
+ CreateRace race = createRace(table -> Table.builder(table)
+ .setParameter(TRINO_QUERY_ID_NAME, "another_query")
+ .withStorage(storage ->
storage.setLocation("local:///other_table"))
+ .build(), false);
+
+ assertThatThrownBy(() -> race.metadata().createTable(SESSION,
race.tableMetadata(), SaveMode.FAIL))
+ .isInstanceOf(TableAlreadyExistsException.class);
+
+ assertThat(race.attemptedTable().get()).isNotNull();
+ assertThat(race.winningTable().get()).isNotNull();
+ assertThat(race.winningTable().get().getStorage().getLocation())
+
.isNotEqualTo(race.attemptedTable().get().getStorage().getLocation());
+ assertThat(metadataExists(race, race.attemptedTable().get()))
+ .isTrue();
+ }
+
+ @Test
+ void testCreateTableRacePreservesMetadataWhenWinnerLookupFails()
+ {
+ CreateRace race =
createRace(TestHudiMetadata::asTableCreatedByAnotherQuery, true);
+
+ assertThatThrownBy(() -> race.metadata().createTable(SESSION,
race.tableMetadata(), SaveMode.FAIL))
+ .isInstanceOf(TableAlreadyExistsException.class)
+ .satisfies(failure ->
assertThat(failure.getSuppressed()).hasSize(1));
+
+ assertThat(race.attemptedTable().get()).isNotNull();
+ assertThat(metadataExists(race, race.attemptedTable().get()))
+ .isTrue();
+ }
+
+ @Test
+ void testCreateTableFailureCleansMetadataWhenNoTableWasRegistered()
+ {
+ CreateRace race = createRace(UnaryOperator.identity(), false, false);
+
+ assertThatThrownBy(() -> race.metadata().createTable(SESSION,
race.tableMetadata(), SaveMode.FAIL))
+ .isInstanceOf(TableAlreadyExistsException.class);
+
+ assertThat(race.attemptedTable().get()).isNotNull();
+ assertThat(race.winningTable().get()).isNull();
+ assertThat(metadataExists(race, race.attemptedTable().get()))
+ .isFalse();
+ }
+
+ @Test
+ void testCreateTableInitializationConflictPreservesExistingMetadata()
+ {
+ SchemaTableName tableName = new SchemaTableName("test_schema",
"concurrent_create");
+ ConnectorTableMetadata tableMetadata = createTableMetadata(tableName);
+ String basePath = "local:///test_schema/concurrent_create";
+ LocalFileSystemFactory localFileSystemFactory = new
LocalFileSystemFactory(temporaryDirectory);
+ TrinoFileSystem localFileSystem =
localFileSystemFactory.create(SESSION);
+ HoodieSchema tableSchema =
HudiSchemaConverter.toTableSchema(tableMetadata.getColumns(),
tableName.getTableName());
+ AtomicBoolean winnerInitialized = new AtomicBoolean();
+ TrinoFileSystem racingFileSystem = (TrinoFileSystem)
Proxy.newProxyInstance(
+ TrinoFileSystem.class.getClassLoader(),
+ new Class<?>[] {TrinoFileSystem.class},
+ (proxy, method, arguments) -> {
+ if (method.getName().equals("listFiles") &&
winnerInitialized.compareAndSet(false, true)) {
+ HudiTableInitializer.initializeTable(localFileSystem,
basePath, tableMetadata, tableSchema);
+ return FileIterator.empty();
+ }
+ try {
+ return method.invoke(localFileSystem, arguments);
+ }
+ catch (InvocationTargetException e) {
+ throw e.getCause();
+ }
+ });
+ TrinoFileSystemFactory racingFileSystemFactory = identity ->
racingFileSystem;
+ HiveMetastore metastore = (HiveMetastore) Proxy.newProxyInstance(
+ HiveMetastore.class.getClassLoader(),
+ new Class<?>[] {HiveMetastore.class},
+ (proxy, method, arguments) -> switch (method.getName()) {
+ case "getDatabase" -> Optional.of(Database.builder()
+ .setDatabaseName(tableName.getSchemaName())
+ .setLocation(Optional.of("local:///test_schema"))
+ .setOwnerName(Optional.of("public"))
+ .setOwnerType(Optional.of(PrincipalType.ROLE))
+ .build());
+ case "getTable" -> Optional.empty();
+ default -> throw new AssertionError("Unexpected metastore
call: " + method);
+ });
+ HudiMetadata metadata = new HudiMetadata(
+ metastore,
+ racingFileSystemFactory,
+ unusedTypeManager(),
+ newDirectExecutorService());
+
+ assertThatThrownBy(() -> metadata.createTable(SESSION, tableMetadata,
SaveMode.FAIL))
+ .isInstanceOf(TrinoException.class)
+ .satisfies(failure -> assertThat(getCausalChain(failure))
+
.anyMatch(FileAlreadyExistsException.class::isInstance));
+
+ assertThat(winnerInitialized).isTrue();
+ assertThat(HudiUtil.hudiMetadataExists(localFileSystem,
Location.of(basePath)))
+ .isTrue();
+ }
+
+ @Test
+ void testDropTreatsExternalTableTypeAsSufficientToPreserveStorage()
+ {
+ SchemaTableName tableName = new SchemaTableName("test_schema",
"external_by_type");
+ ConnectorTableMetadata tableMetadata = createTableMetadata(tableName);
+ String basePath = "local:///test_schema/external_by_type";
+ LocalFileSystemFactory fileSystemFactory = new
LocalFileSystemFactory(temporaryDirectory);
+ TrinoFileSystem fileSystem = fileSystemFactory.create(SESSION);
+ HoodieSchema tableSchema = HudiSchemaConverter.toTableSchema(
+ tableMetadata.getColumns(), tableName.getTableName());
+ HudiTableInitializer.initializeTable(fileSystem, basePath,
tableMetadata, tableSchema);
+
+ Table table = Table.builder(HudiMetastoreTables.buildTable(
+ tableName,
+ basePath,
+ HoodieTableType.COPY_ON_WRITE,
+ tableSchema,
+ List.of(),
+ false,
+ Optional.empty(),
+ Optional.empty()))
+ // Some metastores use only the type signal for external
tables.
+ .setTableType(EXTERNAL_TABLE.name())
Review Comment:
**major:** This passes even with `isExternalTable` removed: `buildTable`
never sets `hudi.trino.managed`, so the marker clause at
`HudiMetadata.java:467` already returns false. Also, no test drops an unmarked
MANAGED table whose name matches `hoodie.table.name`, which is what Spark SQL
CREATE and hive-sync with `create_managed_table=true` register (HUDI-5787
#7940); removing the marker clause fails no test.
Could we set the marker here, and add an unmarked MANAGED case asserting
`deleteData == false` and that a data file survives?
```suggestion
.setTableType(EXTERNAL_TABLE.name())
.setParameter(HudiMetadata.TRINO_MANAGED_TABLE_PARAMETER,
"true")
```
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiTableInitializer.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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 io.trino.filesystem.TrinoFileSystem;
+import io.trino.plugin.hudi.storage.TrinoStorageConfiguration;
+import io.trino.spi.TrinoException;
+import io.trino.spi.connector.ConnectorTableMetadata;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaUtils;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.HoodieTableVersion;
+import org.apache.hudi.keygen.constant.KeyGeneratorType;
+import org.apache.hudi.storage.StoragePath;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import static io.trino.plugin.hudi.HudiErrorCode.HUDI_META_CLIENT_ERROR;
+import static
io.trino.plugin.hudi.HudiTableProperties.getHiveStylePartitioning;
+import static io.trino.plugin.hudi.HudiTableProperties.getHoodieProperties;
+import static io.trino.plugin.hudi.HudiTableProperties.getKeyGeneratorClass;
+import static io.trino.plugin.hudi.HudiTableProperties.getOrderingFields;
+import static io.trino.plugin.hudi.HudiTableProperties.getPartitionedBy;
+import static io.trino.plugin.hudi.HudiTableProperties.getPrimaryKey;
+import static io.trino.plugin.hudi.HudiTableProperties.getRecordMergeMode;
+import static io.trino.plugin.hudi.HudiTableProperties.getTableType;
+import static java.lang.String.format;
+
+/**
+ * Writes a table's {@code .hoodie} directory. Pure hudi-common: no Spark, no
write client and no
+ * metastore, so it can run before the catalog entry exists without entering
any row-writing path.
+ */
+public final class HudiTableInitializer
+{
+ /**
+ * The {@code hoodie.table.version} written by this connector.
+ * <p>
+ * Pinned rather than inherited from {@link HoodieTableVersion#current()}
so that a future bump of
+ * {@code current()} cannot silently change what the connector emits.
Raising this should be a
+ * deliberate change with a test diff, not a side effect of upgrading
Hudi; the literal assertion
+ * in {@code TestHudiTableInitializer} guards this value.
+ */
+ public static final HoodieTableVersion CREATED_TABLE_VERSION =
HoodieTableVersion.TEN;
+
+ private HudiTableInitializer() {}
+
+ /**
+ * Initializes storage for an empty table.
+ *
+ * @param tableSchema the schema that also produces the metastore column
list; Hudi metadata
+ * fields are removed only from the persisted table create schema
+ */
+ public static void initializeTable(
+ TrinoFileSystem fileSystem,
+ String basePath,
+ ConnectorTableMetadata tableMetadata,
+ HoodieSchema tableSchema)
+ {
+ Map<String, Object> properties = tableMetadata.getProperties();
+ HoodieTableMetaClient.TableBuilder builder =
HoodieTableMetaClient.newTableBuilder()
+ .setTableType(getTableType(properties))
+ .setTableName(tableMetadata.getTable().getTableName())
+ .setDatabaseName(tableMetadata.getTable().getSchemaName())
+ .setTableVersion(CREATED_TABLE_VERSION)
+
.setTableCreateSchema(HoodieSchemaUtils.removeMetadataFields(tableSchema).toAvroSchema().toString());
+
+ List<String> primaryKey = getPrimaryKey(properties);
+ if (!primaryKey.isEmpty()) {
+ builder.setRecordKeyFields(String.join(",", primaryKey));
+ }
+ List<String> partitionedBy = getPartitionedBy(properties);
+ if (!partitionedBy.isEmpty()) {
+ builder.setPartitionFields(String.join(",", partitionedBy));
+ }
+ List<String> orderingFields = getOrderingFields(properties);
+ if (!orderingFields.isEmpty()) {
+ builder.setOrderingFields(String.join(",", orderingFields));
+ }
+ getRecordMergeMode(properties).ifPresent(builder::setRecordMergeMode);
+ getKeyGeneratorClass(properties).ifPresentOrElse(
+ builder::setKeyGeneratorClassProp,
+ () ->
builder.setKeyGeneratorType(inferKeyGeneratorType(primaryKey,
partitionedBy).name()));
+
builder.setHiveStylePartitioningEnable(getHiveStylePartitioning(properties));
+
+ Map<String, String> passthrough = getHoodieProperties(properties);
+ if (!passthrough.isEmpty()) {
+ // Applied last so an explicit hoodie.* config wins over anything
derived above; that is
+ // the escape hatch the passthrough map exists for.
+ builder.set(Map.copyOf(passthrough));
+ }
+
+ try {
+ // Goes through Hudi's pluggable-storage extension point: the
configuration names
+ // HudiTrinoStorage as HOODIE_STORAGE_CLASS and carries the
session's file system, so
+ // hudi-common builds storage for the connector itself. No
hudi-common change needed.
+ builder.initTable(new TrinoStorageConfiguration(fileSystem), new
StoragePath(basePath));
+ }
+ catch (IOException e) {
+ throw new TrinoException(HUDI_META_CLIENT_ERROR, format(
+ "Failed to initialize Hudi table metadata for %s at %s",
+ tableMetadata.getTable(), basePath), e);
+ }
+ }
+
+ /**
+ * Mirrors {@code KeyGenUtils.inferKeyGeneratorType} for Trino's plain
column-name lists without
+ * taking a runtime dependency on hudi-client-common.
+ */
+ private static KeyGeneratorType inferKeyGeneratorType(List<String>
primaryKey, List<String> partitionedBy)
+ {
+ if (partitionedBy.isEmpty()) {
+ return KeyGeneratorType.NON_PARTITION;
+ }
+ if (partitionedBy.size() == 1 && primaryKey.size() <= 1) {
+ return KeyGeneratorType.SIMPLE;
+ }
+ return KeyGeneratorType.COMPLEX;
Review Comment:
**major:** The COMPLEX branch has no test; `TestHudiDdl` pins only SIMPLE
and NON_PARTITION. The Flink catalog shipped this exact bug (wrong key
generator for multi-key / multi-partition CREATE, HUDI-7492 #10840), and a
wrong type here changes the record-key encoding every later writer uses.
Could `TestHudiTableInitializer` add a parameterized case comparing the
persisted type with `KeyGenUtils.inferKeyGeneratorType` for {0,1,2} keys x
{0,1,2} partition fields? hudi-client-common is already on the hudi-trino-tests
classpath.
##########
hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java:
##########
@@ -744,6 +746,31 @@ public void testSyncCOWTableWithProperties(boolean
useSchemaFromCommitMetadata,
reInitHiveSyncClient();
reSyncHiveTable();
+ if (syncAsDataSourceTable) {
+ Map<String, String> syncedDescriptorParameters;
+ IMetaStoreClient client = IMetaStoreClientUtil.getMSC(getHiveConf());
+ try {
+ syncedDescriptorParameters = client.getTable(HiveTestUtil.DB_NAME,
HiveTestUtil.TABLE_NAME)
+ .getParameters()
+ .entrySet()
+ .stream()
+ .filter(entry -> entry.getKey().equals("EXTERNAL") ||
entry.getKey().startsWith("spark.sql."))
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+ } finally {
+ client.close();
+ }
+
+ assertEquals(
Review Comment:
**minor:** This parity holds only because `simple-test.avsc` has no field
docs and `HIVE_SYNC_COMMENT` is off here, while `forSnapshotView` now always
sets `includeFieldDocs(true)`. A divergence in how comments land in
`spark.sql.sources.schema.*` would not be caught.
Not blocking, but could the comment-sync test (around line 1233,
`HIVE_SYNC_COMMENT=true`) run the same assertion?
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMetadata.java:
##########
@@ -0,0 +1,412 @@
+/*
+ * 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 io.trino.filesystem.FileIterator;
+import io.trino.filesystem.Location;
+import io.trino.filesystem.TrinoFileSystem;
+import io.trino.filesystem.TrinoFileSystemFactory;
+import io.trino.filesystem.local.LocalFileSystemFactory;
+import io.trino.metastore.Database;
+import io.trino.metastore.HiveMetastore;
+import io.trino.metastore.Table;
+import io.trino.metastore.TableAlreadyExistsException;
+import io.trino.plugin.hudi.util.HudiSchemaConverter;
+import io.trino.spi.TrinoException;
+import io.trino.spi.connector.ColumnMetadata;
+import io.trino.spi.connector.ConnectorTableMetadata;
+import io.trino.spi.connector.SaveMode;
+import io.trino.spi.connector.SchemaTableName;
+import io.trino.spi.predicate.TupleDomain;
+import io.trino.spi.security.PrincipalType;
+import io.trino.spi.type.TypeManager;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Proxy;
+import java.nio.file.FileAlreadyExistsException;
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.OptionalLong;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.UnaryOperator;
+
+import static com.google.common.base.Throwables.getCausalChain;
+import static
com.google.common.util.concurrent.MoreExecutors.newDirectExecutorService;
+import static io.trino.plugin.hive.HiveMetadata.TRINO_QUERY_ID_NAME;
+import static io.trino.plugin.hive.TableType.EXTERNAL_TABLE;
+import static io.trino.plugin.hudi.HudiTableProperties.TABLE_TYPE_PROPERTY;
+import static io.trino.spi.type.BigintType.BIGINT;
+import static io.trino.testing.TestingConnectorSession.SESSION;
+import static org.apache.hudi.common.model.HoodieTableType.MERGE_ON_READ;
+import static org.apache.hudi.common.util.ConfigUtils.IS_QUERY_AS_RO_TABLE;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+final class TestHudiMetadata
+{
+ @TempDir
+ private Path temporaryDirectory;
+
+ @Test
+ void testCreateTableRaceDoesNotDeleteWinningTableMetadata()
+ {
+ CreateRace race =
createRace(TestHudiMetadata::asTableCreatedByAnotherQuery, false);
+
+ assertThatThrownBy(() -> race.metadata().createTable(SESSION,
race.tableMetadata(), SaveMode.FAIL))
+ .isInstanceOf(TableAlreadyExistsException.class);
+
+ assertThat(race.winningTable().get()).isNotNull();
+ assertThat(metadataExists(race, race.winningTable().get()))
+ .isTrue();
+ }
+
+ @Test
+ void
testCreateTableRetryRecognizesCommittedTableAfterLocationNormalization()
+ {
+ CreateRace race = createRace(table -> Table.builder(table)
+ .withStorage(storage -> storage.setLocation(
+ table.getStorage().getLocation().replace("local:///",
"local:/")))
+ .build(), false);
+
+ assertThatCode(() -> race.metadata().createTable(SESSION,
race.tableMetadata(), SaveMode.FAIL))
+ .doesNotThrowAnyException();
+
+ assertThat(race.attemptedTable().get()).isNotNull();
+ assertThat(race.attemptedTable().get().getParameters())
+ .containsEntry(TRINO_QUERY_ID_NAME, SESSION.getQueryId())
+ .containsEntry(HudiMetadata.TRINO_MANAGED_TABLE_PARAMETER,
"true");
+ assertThat(metadataExists(race, race.attemptedTable().get()))
+ .isTrue();
+ }
+
+ @Test
+ void testCreateTableRacePreservesMetadataWhenWinnerUsesDifferentLocation()
+ {
+ CreateRace race = createRace(table -> Table.builder(table)
+ .setParameter(TRINO_QUERY_ID_NAME, "another_query")
+ .withStorage(storage ->
storage.setLocation("local:///other_table"))
+ .build(), false);
+
+ assertThatThrownBy(() -> race.metadata().createTable(SESSION,
race.tableMetadata(), SaveMode.FAIL))
+ .isInstanceOf(TableAlreadyExistsException.class);
+
+ assertThat(race.attemptedTable().get()).isNotNull();
+ assertThat(race.winningTable().get()).isNotNull();
+ assertThat(race.winningTable().get().getStorage().getLocation())
+
.isNotEqualTo(race.attemptedTable().get().getStorage().getLocation());
+ assertThat(metadataExists(race, race.attemptedTable().get()))
+ .isTrue();
+ }
+
+ @Test
+ void testCreateTableRacePreservesMetadataWhenWinnerLookupFails()
+ {
+ CreateRace race =
createRace(TestHudiMetadata::asTableCreatedByAnotherQuery, true);
+
+ assertThatThrownBy(() -> race.metadata().createTable(SESSION,
race.tableMetadata(), SaveMode.FAIL))
+ .isInstanceOf(TableAlreadyExistsException.class)
+ .satisfies(failure ->
assertThat(failure.getSuppressed()).hasSize(1));
+
+ assertThat(race.attemptedTable().get()).isNotNull();
+ assertThat(metadataExists(race, race.attemptedTable().get()))
+ .isTrue();
+ }
+
+ @Test
+ void testCreateTableFailureCleansMetadataWhenNoTableWasRegistered()
+ {
+ CreateRace race = createRace(UnaryOperator.identity(), false, false);
+
+ assertThatThrownBy(() -> race.metadata().createTable(SESSION,
race.tableMetadata(), SaveMode.FAIL))
+ .isInstanceOf(TableAlreadyExistsException.class);
+
+ assertThat(race.attemptedTable().get()).isNotNull();
+ assertThat(race.winningTable().get()).isNull();
+ assertThat(metadataExists(race, race.attemptedTable().get()))
+ .isFalse();
+ }
+
+ @Test
+ void testCreateTableInitializationConflictPreservesExistingMetadata()
+ {
+ SchemaTableName tableName = new SchemaTableName("test_schema",
"concurrent_create");
+ ConnectorTableMetadata tableMetadata = createTableMetadata(tableName);
+ String basePath = "local:///test_schema/concurrent_create";
+ LocalFileSystemFactory localFileSystemFactory = new
LocalFileSystemFactory(temporaryDirectory);
+ TrinoFileSystem localFileSystem =
localFileSystemFactory.create(SESSION);
+ HoodieSchema tableSchema =
HudiSchemaConverter.toTableSchema(tableMetadata.getColumns(),
tableName.getTableName());
+ AtomicBoolean winnerInitialized = new AtomicBoolean();
+ TrinoFileSystem racingFileSystem = (TrinoFileSystem)
Proxy.newProxyInstance(
+ TrinoFileSystem.class.getClassLoader(),
+ new Class<?>[] {TrinoFileSystem.class},
+ (proxy, method, arguments) -> {
+ if (method.getName().equals("listFiles") &&
winnerInitialized.compareAndSet(false, true)) {
+ HudiTableInitializer.initializeTable(localFileSystem,
basePath, tableMetadata, tableSchema);
+ return FileIterator.empty();
+ }
+ try {
+ return method.invoke(localFileSystem, arguments);
+ }
+ catch (InvocationTargetException e) {
+ throw e.getCause();
+ }
+ });
+ TrinoFileSystemFactory racingFileSystemFactory = identity ->
racingFileSystem;
+ HiveMetastore metastore = (HiveMetastore) Proxy.newProxyInstance(
+ HiveMetastore.class.getClassLoader(),
+ new Class<?>[] {HiveMetastore.class},
+ (proxy, method, arguments) -> switch (method.getName()) {
+ case "getDatabase" -> Optional.of(Database.builder()
+ .setDatabaseName(tableName.getSchemaName())
+ .setLocation(Optional.of("local:///test_schema"))
+ .setOwnerName(Optional.of("public"))
+ .setOwnerType(Optional.of(PrincipalType.ROLE))
+ .build());
+ case "getTable" -> Optional.empty();
+ default -> throw new AssertionError("Unexpected metastore
call: " + method);
+ });
+ HudiMetadata metadata = new HudiMetadata(
+ metastore,
+ racingFileSystemFactory,
+ unusedTypeManager(),
+ newDirectExecutorService());
+
+ assertThatThrownBy(() -> metadata.createTable(SESSION, tableMetadata,
SaveMode.FAIL))
+ .isInstanceOf(TrinoException.class)
+ .satisfies(failure -> assertThat(getCausalChain(failure))
+
.anyMatch(FileAlreadyExistsException.class::isInstance));
+
+ assertThat(winnerInitialized).isTrue();
+ assertThat(HudiUtil.hudiMetadataExists(localFileSystem,
Location.of(basePath)))
+ .isTrue();
+ }
+
+ @Test
+ void testDropTreatsExternalTableTypeAsSufficientToPreserveStorage()
+ {
+ SchemaTableName tableName = new SchemaTableName("test_schema",
"external_by_type");
+ ConnectorTableMetadata tableMetadata = createTableMetadata(tableName);
+ String basePath = "local:///test_schema/external_by_type";
+ LocalFileSystemFactory fileSystemFactory = new
LocalFileSystemFactory(temporaryDirectory);
+ TrinoFileSystem fileSystem = fileSystemFactory.create(SESSION);
+ HoodieSchema tableSchema = HudiSchemaConverter.toTableSchema(
+ tableMetadata.getColumns(), tableName.getTableName());
+ HudiTableInitializer.initializeTable(fileSystem, basePath,
tableMetadata, tableSchema);
+
+ Table table = Table.builder(HudiMetastoreTables.buildTable(
+ tableName,
+ basePath,
+ HoodieTableType.COPY_ON_WRITE,
+ tableSchema,
+ List.of(),
+ false,
+ Optional.empty(),
+ Optional.empty()))
+ // Some metastores use only the type signal for external
tables.
+ .setTableType(EXTERNAL_TABLE.name())
+ .build();
+ AtomicReference<Boolean> deleteData = new AtomicReference<>();
+ HiveMetastore metastore = (HiveMetastore) Proxy.newProxyInstance(
Review Comment:
**minor:** This class hand-writes six
`Proxy.newProxyInstance(HiveMetastore...)` switch blocks (plus one in
`TestHudiRegisterTableProcedure`), three copies of the initialize-table setup
(around lines 165, 215, 277) and two eleven-argument `HudiTableHandle`
constructors (246, 313).
Not blocking, but could we pull these into a `fakeMetastore(Map<String,
Function<Object[], Object>>)` helper plus `initializedTable(...)` /
`tableHandle(...)` helpers?
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMetadata.java:
##########
@@ -0,0 +1,412 @@
+/*
+ * 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 io.trino.filesystem.FileIterator;
+import io.trino.filesystem.Location;
+import io.trino.filesystem.TrinoFileSystem;
+import io.trino.filesystem.TrinoFileSystemFactory;
+import io.trino.filesystem.local.LocalFileSystemFactory;
+import io.trino.metastore.Database;
+import io.trino.metastore.HiveMetastore;
+import io.trino.metastore.Table;
+import io.trino.metastore.TableAlreadyExistsException;
+import io.trino.plugin.hudi.util.HudiSchemaConverter;
+import io.trino.spi.TrinoException;
+import io.trino.spi.connector.ColumnMetadata;
+import io.trino.spi.connector.ConnectorTableMetadata;
+import io.trino.spi.connector.SaveMode;
+import io.trino.spi.connector.SchemaTableName;
+import io.trino.spi.predicate.TupleDomain;
+import io.trino.spi.security.PrincipalType;
+import io.trino.spi.type.TypeManager;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Proxy;
+import java.nio.file.FileAlreadyExistsException;
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.OptionalLong;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.UnaryOperator;
+
+import static com.google.common.base.Throwables.getCausalChain;
+import static
com.google.common.util.concurrent.MoreExecutors.newDirectExecutorService;
+import static io.trino.plugin.hive.HiveMetadata.TRINO_QUERY_ID_NAME;
+import static io.trino.plugin.hive.TableType.EXTERNAL_TABLE;
+import static io.trino.plugin.hudi.HudiTableProperties.TABLE_TYPE_PROPERTY;
+import static io.trino.spi.type.BigintType.BIGINT;
+import static io.trino.testing.TestingConnectorSession.SESSION;
+import static org.apache.hudi.common.model.HoodieTableType.MERGE_ON_READ;
+import static org.apache.hudi.common.util.ConfigUtils.IS_QUERY_AS_RO_TABLE;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+final class TestHudiMetadata
+{
+ @TempDir
+ private Path temporaryDirectory;
+
+ @Test
+ void testCreateTableRaceDoesNotDeleteWinningTableMetadata()
+ {
+ CreateRace race =
createRace(TestHudiMetadata::asTableCreatedByAnotherQuery, false);
+
+ assertThatThrownBy(() -> race.metadata().createTable(SESSION,
race.tableMetadata(), SaveMode.FAIL))
+ .isInstanceOf(TableAlreadyExistsException.class);
+
+ assertThat(race.winningTable().get()).isNotNull();
+ assertThat(metadataExists(race, race.winningTable().get()))
+ .isTrue();
+ }
+
+ @Test
+ void
testCreateTableRetryRecognizesCommittedTableAfterLocationNormalization()
+ {
+ CreateRace race = createRace(table -> Table.builder(table)
+ .withStorage(storage -> storage.setLocation(
+ table.getStorage().getLocation().replace("local:///",
"local:/")))
+ .build(), false);
+
+ assertThatCode(() -> race.metadata().createTable(SESSION,
race.tableMetadata(), SaveMode.FAIL))
+ .doesNotThrowAnyException();
+
+ assertThat(race.attemptedTable().get()).isNotNull();
+ assertThat(race.attemptedTable().get().getParameters())
+ .containsEntry(TRINO_QUERY_ID_NAME, SESSION.getQueryId())
+ .containsEntry(HudiMetadata.TRINO_MANAGED_TABLE_PARAMETER,
"true");
+ assertThat(metadataExists(race, race.attemptedTable().get()))
+ .isTrue();
+ }
+
+ @Test
+ void testCreateTableRacePreservesMetadataWhenWinnerUsesDifferentLocation()
+ {
+ CreateRace race = createRace(table -> Table.builder(table)
+ .setParameter(TRINO_QUERY_ID_NAME, "another_query")
+ .withStorage(storage ->
storage.setLocation("local:///other_table"))
+ .build(), false);
+
+ assertThatThrownBy(() -> race.metadata().createTable(SESSION,
race.tableMetadata(), SaveMode.FAIL))
+ .isInstanceOf(TableAlreadyExistsException.class);
+
+ assertThat(race.attemptedTable().get()).isNotNull();
+ assertThat(race.winningTable().get()).isNotNull();
+ assertThat(race.winningTable().get().getStorage().getLocation())
+
.isNotEqualTo(race.attemptedTable().get().getStorage().getLocation());
+ assertThat(metadataExists(race, race.attemptedTable().get()))
+ .isTrue();
+ }
+
+ @Test
+ void testCreateTableRacePreservesMetadataWhenWinnerLookupFails()
+ {
+ CreateRace race =
createRace(TestHudiMetadata::asTableCreatedByAnotherQuery, true);
+
+ assertThatThrownBy(() -> race.metadata().createTable(SESSION,
race.tableMetadata(), SaveMode.FAIL))
+ .isInstanceOf(TableAlreadyExistsException.class)
+ .satisfies(failure ->
assertThat(failure.getSuppressed()).hasSize(1));
+
+ assertThat(race.attemptedTable().get()).isNotNull();
+ assertThat(metadataExists(race, race.attemptedTable().get()))
+ .isTrue();
+ }
+
+ @Test
+ void testCreateTableFailureCleansMetadataWhenNoTableWasRegistered()
+ {
+ CreateRace race = createRace(UnaryOperator.identity(), false, false);
+
+ assertThatThrownBy(() -> race.metadata().createTable(SESSION,
race.tableMetadata(), SaveMode.FAIL))
+ .isInstanceOf(TableAlreadyExistsException.class);
+
+ assertThat(race.attemptedTable().get()).isNotNull();
+ assertThat(race.winningTable().get()).isNull();
+ assertThat(metadataExists(race, race.attemptedTable().get()))
+ .isFalse();
+ }
+
+ @Test
+ void testCreateTableInitializationConflictPreservesExistingMetadata()
+ {
+ SchemaTableName tableName = new SchemaTableName("test_schema",
"concurrent_create");
+ ConnectorTableMetadata tableMetadata = createTableMetadata(tableName);
+ String basePath = "local:///test_schema/concurrent_create";
+ LocalFileSystemFactory localFileSystemFactory = new
LocalFileSystemFactory(temporaryDirectory);
+ TrinoFileSystem localFileSystem =
localFileSystemFactory.create(SESSION);
+ HoodieSchema tableSchema =
HudiSchemaConverter.toTableSchema(tableMetadata.getColumns(),
tableName.getTableName());
+ AtomicBoolean winnerInitialized = new AtomicBoolean();
+ TrinoFileSystem racingFileSystem = (TrinoFileSystem)
Proxy.newProxyInstance(
+ TrinoFileSystem.class.getClassLoader(),
+ new Class<?>[] {TrinoFileSystem.class},
+ (proxy, method, arguments) -> {
+ if (method.getName().equals("listFiles") &&
winnerInitialized.compareAndSet(false, true)) {
+ HudiTableInitializer.initializeTable(localFileSystem,
basePath, tableMetadata, tableSchema);
+ return FileIterator.empty();
+ }
+ try {
+ return method.invoke(localFileSystem, arguments);
+ }
+ catch (InvocationTargetException e) {
+ throw e.getCause();
+ }
+ });
+ TrinoFileSystemFactory racingFileSystemFactory = identity ->
racingFileSystem;
+ HiveMetastore metastore = (HiveMetastore) Proxy.newProxyInstance(
+ HiveMetastore.class.getClassLoader(),
+ new Class<?>[] {HiveMetastore.class},
+ (proxy, method, arguments) -> switch (method.getName()) {
+ case "getDatabase" -> Optional.of(Database.builder()
+ .setDatabaseName(tableName.getSchemaName())
+ .setLocation(Optional.of("local:///test_schema"))
+ .setOwnerName(Optional.of("public"))
+ .setOwnerType(Optional.of(PrincipalType.ROLE))
+ .build());
+ case "getTable" -> Optional.empty();
+ default -> throw new AssertionError("Unexpected metastore
call: " + method);
+ });
+ HudiMetadata metadata = new HudiMetadata(
+ metastore,
+ racingFileSystemFactory,
+ unusedTypeManager(),
+ newDirectExecutorService());
+
+ assertThatThrownBy(() -> metadata.createTable(SESSION, tableMetadata,
SaveMode.FAIL))
+ .isInstanceOf(TrinoException.class)
+ .satisfies(failure -> assertThat(getCausalChain(failure))
+
.anyMatch(FileAlreadyExistsException.class::isInstance));
+
+ assertThat(winnerInitialized).isTrue();
+ assertThat(HudiUtil.hudiMetadataExists(localFileSystem,
Location.of(basePath)))
+ .isTrue();
+ }
+
+ @Test
+ void testDropTreatsExternalTableTypeAsSufficientToPreserveStorage()
+ {
+ SchemaTableName tableName = new SchemaTableName("test_schema",
"external_by_type");
+ ConnectorTableMetadata tableMetadata = createTableMetadata(tableName);
+ String basePath = "local:///test_schema/external_by_type";
+ LocalFileSystemFactory fileSystemFactory = new
LocalFileSystemFactory(temporaryDirectory);
+ TrinoFileSystem fileSystem = fileSystemFactory.create(SESSION);
+ HoodieSchema tableSchema = HudiSchemaConverter.toTableSchema(
+ tableMetadata.getColumns(), tableName.getTableName());
+ HudiTableInitializer.initializeTable(fileSystem, basePath,
tableMetadata, tableSchema);
+
+ Table table = Table.builder(HudiMetastoreTables.buildTable(
+ tableName,
+ basePath,
+ HoodieTableType.COPY_ON_WRITE,
+ tableSchema,
+ List.of(),
+ false,
+ Optional.empty(),
+ Optional.empty()))
+ // Some metastores use only the type signal for external
tables.
+ .setTableType(EXTERNAL_TABLE.name())
+ .build();
+ AtomicReference<Boolean> deleteData = new AtomicReference<>();
+ HiveMetastore metastore = (HiveMetastore) Proxy.newProxyInstance(
+ HiveMetastore.class.getClassLoader(),
+ new Class<?>[] {HiveMetastore.class},
+ (proxy, method, arguments) -> switch (method.getName()) {
+ case "getTable" -> Optional.of(table);
+ case "dropTable" -> {
+ deleteData.set((Boolean) arguments[2]);
+ yield null;
+ }
+ default -> throw new AssertionError("Unexpected metastore
call: " + method);
+ });
+ HudiMetadata metadata = new HudiMetadata(
+ metastore,
+ fileSystemFactory,
+ unusedTypeManager(),
+ newDirectExecutorService());
+ HudiTableHandle tableHandle = new HudiTableHandle(
+ tableName.getSchemaName(),
+ tableName.getTableName(),
+ basePath,
+ HoodieTableType.COPY_ON_WRITE,
+ List.of(),
+ List.of(),
+ TupleDomain.all(),
+ TupleDomain.all(),
+ OptionalLong.empty(),
+ tableSchema.toAvroSchema().toString(),
+ "0");
+
+ metadata.dropTable(SESSION, tableHandle);
+
+ assertThat(table.getParameters()).doesNotContainKey("EXTERNAL");
+ assertThat(deleteData).hasValue(false);
+ assertThat(HudiUtil.hudiMetadataExists(fileSystem,
Location.of(basePath))).isTrue();
+ }
+
+ @Test
+ void testDropManagedHiveSyncViewsPreservesSharedStorage()
+ throws IOException
+ {
+ String basePath = "local:///test_schema/shared_mor";
+ LocalFileSystemFactory fileSystemFactory = new
LocalFileSystemFactory(temporaryDirectory);
+ TrinoFileSystem fileSystem = fileSystemFactory.create(SESSION);
+ SchemaTableName physicalName = new SchemaTableName("test_schema",
"shared_mor");
+ ConnectorTableMetadata tableMetadata = new ConnectorTableMetadata(
+ physicalName, createTableMetadata(physicalName).getColumns(),
Map.of(TABLE_TYPE_PROPERTY, MERGE_ON_READ));
+ HoodieSchema tableSchema =
HudiSchemaConverter.toTableSchema(tableMetadata.getColumns(),
physicalName.getTableName());
+ HudiTableInitializer.initializeTable(fileSystem, basePath,
tableMetadata, tableSchema);
+ Location dataFile =
Location.of(basePath).appendPath("existing.parquet");
+ fileSystem.newOutputFile(dataFile).createOrOverwrite(new byte[] {1});
+
+ Map<String, Table> tables = new HashMap<>();
+ for (String viewName : List.of("shared_mor_ro", "shared_mor_rt")) {
+ SchemaTableName name = new SchemaTableName("test_schema",
viewName);
+ Table table = HudiMetastoreTables.buildTable(name, basePath,
MERGE_ON_READ, tableSchema,
+ List.of(), false, Optional.empty(), Optional.empty());
+ tables.put(viewName, table);
+ }
+ // Even a copied ownership marker must not let a read-optimized alias
delete the base path.
+ tables.put("shared_mor_ro", Table.builder(tables.get("shared_mor_ro"))
+ .setParameter(HudiMetadata.TRINO_MANAGED_TABLE_PARAMETER,
"true")
+ .withStorage(storage ->
storage.setSerdeParameters(Map.of(IS_QUERY_AS_RO_TABLE, "true")))
Review Comment:
**minor:** The RO-serde clause (`HudiMetadata.java:468`) never decides this
test: `shared_mor_ro` already fails the `hoodie.table.name` match, so removing
the clause still passes. The flag only matters with `skip_ro_suffix=true`,
where the RO view takes the base name.
Not blocking, but could the RO view here be named `shared_mor`?
--
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]
