voonhous commented on code in PR #20086: URL: https://github.com/apache/hudi/pull/20086#discussion_r4178153498
########## hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiWorkerTableMetadataAccess.java: ########## @@ -0,0 +1,129 @@ +/* + * 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.ImmutableSet; +import io.opentelemetry.sdk.trace.data.SpanData; +import io.trino.metadata.Metadata; +import io.trino.metadata.QualifiedObjectName; +import io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer; +import io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable; +import io.trino.testing.AbstractTestQueryFramework; +import io.trino.testing.DistributedQueryRunner; +import io.trino.testing.QueryRunner; +import org.apache.hudi.common.table.HoodieTableConfig; +import org.apache.hudi.common.table.HoodieTableVersion; +import org.junit.jupiter.api.parallel.Execution; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; + +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.Function; + +import static com.google.common.collect.ImmutableList.toImmutableList; +import static com.google.common.collect.ImmutableMap.toImmutableMap; +import static io.trino.filesystem.tracing.FileSystemAttributes.FILE_LOCATION; +import static io.trino.testing.TransactionBuilder.transaction; +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.parallel.ExecutionMode.SAME_THREAD; + +/** + * Asserts that reading file groups with log files on a worker does not touch the table metadata under + * {@code .hoodie}: the table config, schema and timeline come from the table handle built on the coordinator. + */ +@Execution(SAME_THREAD) +final class TestHudiWorkerTableMetadataAccess + extends AbstractTestQueryFramework +{ + // Spans the task executors open around split processing; file system spans under them come from the page source path + private static final Set<String> SPLIT_PROCESSING_SPANS = ImmutableSet.of("process", "split"); + + @Override + protected QueryRunner createQueryRunner() + throws Exception + { + return HudiQueryRunner.builder() + .addConnectorProperty("fs.cache.enabled", "false") Review Comment: **nit:** `hudi.metadata.cache.enabled` defaults to true, so a cached `.hoodie` timeline read may leave no `FILE_LOCATION` span and slip past the `noneMatch("/.hoodie")` check. `TestHudiNoCacheFileOperations` turns off both caches for that reason. Feel free to ignore, but could we turn it off here too? ```suggestion .addConnectorProperty("fs.cache.enabled", "false") .addConnectorProperty("hudi.metadata.cache.enabled", "false") ``` ########## hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiTableHandle.java: ########## @@ -0,0 +1,114 @@ +/* + * 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.airlift.json.JsonCodec; +import io.trino.spi.predicate.TupleDomain; +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.table.HoodieTableVersion; +import org.apache.hudi.common.table.read.FileGroupReaderTableState; +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.table.timeline.HoodieTimeline; +import org.apache.hudi.common.table.timeline.InstantGenerator; +import org.apache.hudi.common.table.timeline.TimelineLayout; +import org.apache.hudi.storage.StoragePath; +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.OptionalLong; + +import static io.airlift.json.JsonCodec.jsonCodec; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +final class TestHudiTableHandle +{ + private static final JsonCodec<HudiTableHandle> CODEC = jsonCodec(HudiTableHandle.class); + private static final Map<String, String> TABLE_CONFIG = ImmutableMap.of( + "hoodie.table.name", "test_table", + "hoodie.table.type", "MERGE_ON_READ", + "hoodie.table.version", "6"); + + @Test + void testWorkerTableStateFromJson() + { + HudiCommittedInstants committedInstants = new HudiCommittedInstants( + ImmutableList.of("20240201000000000", "20240401000000000"), + ImmutableList.of("20240301000000000"), + Optional.of("20240201000000000")); + HudiTableHandle handle = CODEC.fromJson(CODEC.toJson(createTableHandle(TABLE_CONFIG, Optional.of(committedInstants)))); + + assertThat(handle.getTableConfig()).isEqualTo(TABLE_CONFIG); + assertThat(handle.getCommittedInstants()).contains(committedInstants); + FileGroupReaderTableState tableState = handle.getFileGroupReaderTableState(); + assertThat(tableState.getBasePath()).isEqualTo(new StoragePath("/test/path")); + assertThat(tableState.getTableConfig().getTableName()).isEqualTo("test_table"); + assertThat(tableState.getTableConfig().getTableVersion()).isEqualTo(HoodieTableVersion.SIX); + // Before the timeline start counts as committed (archived), inflight and unknown instants do not + assertThat(tableState.isCommitted("20240101000000000")).isTrue(); + assertThat(tableState.isCommitted("20240401000000000")).isTrue(); + assertThat(tableState.isCommitted("20240301000000000")).isFalse(); + assertThat(tableState.isCommitted("20240215000000000")).isFalse(); + } + + @Test + void testWorkerTableStateRequiresTableConfig() + { + HudiTableHandle handle = CODEC.fromJson(CODEC.toJson(createTableHandle(ImmutableMap.of(), Optional.empty()))); + + assertThatThrownBy(handle::getFileGroupReaderTableState) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("carries no table config"); + } + + @Test + void testCaptureCommittedInstantsUpToLatestCommit() + { + InstantGenerator instantGenerator = TimelineLayout.TIMELINE_LAYOUT_V1.getInstantGenerator(); + List<HoodieInstant> instants = ImmutableList.of( + instantGenerator.createNewInstant(HoodieInstant.State.COMPLETED, HoodieTimeline.DELTA_COMMIT_ACTION, "20240201000000000"), + instantGenerator.createNewInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.DELTA_COMMIT_ACTION, "20240301000000000"), + instantGenerator.createNewInstant(HoodieInstant.State.COMPLETED, HoodieTimeline.DELTA_COMMIT_ACTION, "20240401000000000"), + instantGenerator.createNewInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.DELTA_COMMIT_ACTION, "20240501000000000"), + instantGenerator.createNewInstant(HoodieInstant.State.COMPLETED, HoodieTimeline.DELTA_COMMIT_ACTION, "20240601000000000")); + HoodieTimeline timeline = TimelineLayout.TIMELINE_LAYOUT_V1.getTimelineFactory().createDefaultTimeline(instants.stream(), null); + + assertThat(HudiCommittedInstants.capture(timeline, "20240401000000000")).isEqualTo(new HudiCommittedInstants( + ImmutableList.of("20240201000000000", "20240401000000000"), + ImmutableList.of("20240301000000000"), + Optional.of("20240201000000000"))); + } + + private static HudiTableHandle createTableHandle(Map<String, String> tableConfig, Optional<HudiCommittedInstants> committedInstants) Review Comment: **nit:** This is the fourth copy of the 13-argument handle construction (also in `TestHudiSplitSource`, `TestHudiSplitFactory` and `TestHudiPartitionInfoLoader`), and this commit had to append `ImmutableMap.of(), Optional.empty()` to the other three. Feel free to ignore, but could we add a `HudiTestUtils.testTableHandle(...)` factory so the next handle field is a one-line change? ########## hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiWorkerTableMetadataAccess.java: ########## @@ -0,0 +1,129 @@ +/* + * 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.ImmutableSet; +import io.opentelemetry.sdk.trace.data.SpanData; +import io.trino.metadata.Metadata; +import io.trino.metadata.QualifiedObjectName; +import io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer; +import io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable; +import io.trino.testing.AbstractTestQueryFramework; +import io.trino.testing.DistributedQueryRunner; +import io.trino.testing.QueryRunner; +import org.apache.hudi.common.table.HoodieTableConfig; +import org.apache.hudi.common.table.HoodieTableVersion; +import org.junit.jupiter.api.parallel.Execution; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; + +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.Function; + +import static com.google.common.collect.ImmutableList.toImmutableList; +import static com.google.common.collect.ImmutableMap.toImmutableMap; +import static io.trino.filesystem.tracing.FileSystemAttributes.FILE_LOCATION; +import static io.trino.testing.TransactionBuilder.transaction; +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.parallel.ExecutionMode.SAME_THREAD; + +/** + * Asserts that reading file groups with log files on a worker does not touch the table metadata under + * {@code .hoodie}: the table config, schema and timeline come from the table handle built on the coordinator. + */ +@Execution(SAME_THREAD) +final class TestHudiWorkerTableMetadataAccess + extends AbstractTestQueryFramework +{ + // Spans the task executors open around split processing; file system spans under them come from the page source path + private static final Set<String> SPLIT_PROCESSING_SPANS = ImmutableSet.of("process", "split"); + + @Override + protected QueryRunner createQueryRunner() Review Comment: **nit:** This starts one more query runner over every resource fixture, with nearly the same setup as `TestHudiNoCacheFileOperations` (fs cache off, table statistics off, `ResourceHudiTablesInitializer`). Feel free to ignore, but could these two methods live there instead, or could a comment say why they need their own runner with a separate worker node? ########## hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiWorkerTableMetadataAccess.java: ########## @@ -0,0 +1,129 @@ +/* + * 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.ImmutableSet; +import io.opentelemetry.sdk.trace.data.SpanData; +import io.trino.metadata.Metadata; +import io.trino.metadata.QualifiedObjectName; +import io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer; +import io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable; +import io.trino.testing.AbstractTestQueryFramework; +import io.trino.testing.DistributedQueryRunner; +import io.trino.testing.QueryRunner; +import org.apache.hudi.common.table.HoodieTableConfig; +import org.apache.hudi.common.table.HoodieTableVersion; +import org.junit.jupiter.api.parallel.Execution; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; + +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.Function; + +import static com.google.common.collect.ImmutableList.toImmutableList; +import static com.google.common.collect.ImmutableMap.toImmutableMap; +import static io.trino.filesystem.tracing.FileSystemAttributes.FILE_LOCATION; +import static io.trino.testing.TransactionBuilder.transaction; +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.parallel.ExecutionMode.SAME_THREAD; + +/** + * Asserts that reading file groups with log files on a worker does not touch the table metadata under + * {@code .hoodie}: the table config, schema and timeline come from the table handle built on the coordinator. + */ +@Execution(SAME_THREAD) +final class TestHudiWorkerTableMetadataAccess + extends AbstractTestQueryFramework +{ + // Spans the task executors open around split processing; file system spans under them come from the page source path + private static final Set<String> SPLIT_PROCESSING_SPANS = ImmutableSet.of("process", "split"); + + @Override + protected QueryRunner createQueryRunner() + throws Exception + { + return HudiQueryRunner.builder() + .addConnectorProperty("fs.cache.enabled", "false") + .addConnectorProperty("hudi.table-statistics-enabled", "false") + .setDataLoader(new ResourceHudiTablesInitializer()) + .setWorkerCount(1) + .build(); + } + + @ParameterizedTest + @EnumSource(value = TestingTable.class, names = {"HUDI_COMPREHENSIVE_TYPES_V6_MOR", "HUDI_COMPREHENSIVE_TYPES_V8_MOR"}) + void testLogFileSplitsDoNotReadTableMetadata(TestingTable table) + { + DistributedQueryRunner queryRunner = getDistributedQueryRunner(); + queryRunner.executeWithPlan(getSession(), "SELECT * FROM " + table.getRtTableName()); + + String tableDirectory = "/" + table.getTableName() + "/"; + List<String> splitFileLocations = splitProcessingFileLocations(queryRunner.getSpans()).stream() + .filter(location -> location.contains(tableDirectory)) + .collect(toImmutableList()); + assertThat(splitFileLocations).anyMatch(location -> location.contains(".log.")); + assertThat(splitFileLocations).noneMatch(location -> location.contains("/.hoodie")); Review Comment: **nit:** `FileOperationUtils.FileType.fromFilePath` already classifies `.log` and `.hoodie` paths. Feel free to ignore, but could we map the locations through it and assert `contains(LOG)` plus none of the metadata types, so a failure names which metadata file was read? `fromFilePath` throws on unknown paths, so the table-directory filter would stay first. ########## hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiWorkerTableMetadataAccess.java: ########## @@ -0,0 +1,129 @@ +/* + * 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.ImmutableSet; +import io.opentelemetry.sdk.trace.data.SpanData; +import io.trino.metadata.Metadata; +import io.trino.metadata.QualifiedObjectName; +import io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer; +import io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer.TestingTable; +import io.trino.testing.AbstractTestQueryFramework; +import io.trino.testing.DistributedQueryRunner; +import io.trino.testing.QueryRunner; +import org.apache.hudi.common.table.HoodieTableConfig; +import org.apache.hudi.common.table.HoodieTableVersion; +import org.junit.jupiter.api.parallel.Execution; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; + +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.Function; + +import static com.google.common.collect.ImmutableList.toImmutableList; +import static com.google.common.collect.ImmutableMap.toImmutableMap; +import static io.trino.filesystem.tracing.FileSystemAttributes.FILE_LOCATION; +import static io.trino.testing.TransactionBuilder.transaction; +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.parallel.ExecutionMode.SAME_THREAD; + +/** + * Asserts that reading file groups with log files on a worker does not touch the table metadata under + * {@code .hoodie}: the table config, schema and timeline come from the table handle built on the coordinator. + */ +@Execution(SAME_THREAD) +final class TestHudiWorkerTableMetadataAccess + extends AbstractTestQueryFramework +{ + // Spans the task executors open around split processing; file system spans under them come from the page source path + private static final Set<String> SPLIT_PROCESSING_SPANS = ImmutableSet.of("process", "split"); + + @Override + protected QueryRunner createQueryRunner() + throws Exception + { + return HudiQueryRunner.builder() + .addConnectorProperty("fs.cache.enabled", "false") + .addConnectorProperty("hudi.table-statistics-enabled", "false") + .setDataLoader(new ResourceHudiTablesInitializer()) + .setWorkerCount(1) + .build(); + } + + @ParameterizedTest + @EnumSource(value = TestingTable.class, names = {"HUDI_COMPREHENSIVE_TYPES_V6_MOR", "HUDI_COMPREHENSIVE_TYPES_V8_MOR"}) + void testLogFileSplitsDoNotReadTableMetadata(TestingTable table) + { + DistributedQueryRunner queryRunner = getDistributedQueryRunner(); + queryRunner.executeWithPlan(getSession(), "SELECT * FROM " + table.getRtTableName()); + + String tableDirectory = "/" + table.getTableName() + "/"; + List<String> splitFileLocations = splitProcessingFileLocations(queryRunner.getSpans()).stream() + .filter(location -> location.contains(tableDirectory)) + .collect(toImmutableList()); + assertThat(splitFileLocations).anyMatch(location -> location.contains(".log.")); + assertThat(splitFileLocations).noneMatch(location -> location.contains("/.hoodie")); + } + + @ParameterizedTest + @EnumSource(value = TestingTable.class, names = {"HUDI_COMPREHENSIVE_TYPES_V6_MOR", "HUDI_COMPREHENSIVE_TYPES_V8_MOR"}) + void testTableHandleCarriesWorkerTableState(TestingTable table) + { + HudiTableHandle handle = coordinatorTableHandle(table.getRtTableName()); + + assertThat(handle.getTableSchemaStr()).isNotEmpty(); + assertThat(handle.getTableConfig()) + .containsKey(HoodieTableConfig.VERSION.key()) + .doesNotContainKey(HoodieTableConfig.CREATE_SCHEMA.key()); + assertThat(handle.getCommittedInstants().isPresent()) + .isEqualTo(handle.getFileGroupReaderTableState().getTableConfig().getTableVersion().lesserThan(HoodieTableVersion.EIGHT)); Review Comment: **nit:** The expected value is derived from the handle under test, and only presence is checked. Feel free to ignore, but could we pass the expectation as a parameter (V6 present, V8 empty) and, for V6, assert that `completedInstants()` equals the fixture's delta commit times and `inflightInstants()` is empty? ########## hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiTableHandle.java: ########## @@ -0,0 +1,114 @@ +/* + * 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.airlift.json.JsonCodec; +import io.trino.spi.predicate.TupleDomain; +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.table.HoodieTableVersion; +import org.apache.hudi.common.table.read.FileGroupReaderTableState; +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.table.timeline.HoodieTimeline; +import org.apache.hudi.common.table.timeline.InstantGenerator; +import org.apache.hudi.common.table.timeline.TimelineLayout; +import org.apache.hudi.storage.StoragePath; +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.OptionalLong; + +import static io.airlift.json.JsonCodec.jsonCodec; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +final class TestHudiTableHandle +{ + private static final JsonCodec<HudiTableHandle> CODEC = jsonCodec(HudiTableHandle.class); + private static final Map<String, String> TABLE_CONFIG = ImmutableMap.of( + "hoodie.table.name", "test_table", + "hoodie.table.type", "MERGE_ON_READ", + "hoodie.table.version", "6"); + + @Test + void testWorkerTableStateFromJson() + { + HudiCommittedInstants committedInstants = new HudiCommittedInstants( + ImmutableList.of("20240201000000000", "20240401000000000"), + ImmutableList.of("20240301000000000"), + Optional.of("20240201000000000")); + HudiTableHandle handle = CODEC.fromJson(CODEC.toJson(createTableHandle(TABLE_CONFIG, Optional.of(committedInstants)))); + + assertThat(handle.getTableConfig()).isEqualTo(TABLE_CONFIG); + assertThat(handle.getCommittedInstants()).contains(committedInstants); + FileGroupReaderTableState tableState = handle.getFileGroupReaderTableState(); + assertThat(tableState.getBasePath()).isEqualTo(new StoragePath("/test/path")); + assertThat(tableState.getTableConfig().getTableName()).isEqualTo("test_table"); + assertThat(tableState.getTableConfig().getTableVersion()).isEqualTo(HoodieTableVersion.SIX); + // Before the timeline start counts as committed (archived), inflight and unknown instants do not + assertThat(tableState.isCommitted("20240101000000000")).isTrue(); + assertThat(tableState.isCommitted("20240401000000000")).isTrue(); + assertThat(tableState.isCommitted("20240301000000000")).isFalse(); + assertThat(tableState.isCommitted("20240215000000000")).isFalse(); Review Comment: **nit:** These four checks repeat `TestFileGroupReaderTableState` from #20078 with the same instants (lines 84-87 and 146-147). The new part here is the JSON round trip, which line 57 already asserts. Feel free to ignore, but could we keep one, e.g. the inflight `20240301000000000` not being committed, as the wiring check and leave the semantics to the hudi-common test? -- 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]
