wombatu-kun commented on code in PR #19456:
URL: https://github.com/apache/hudi/pull/19456#discussion_r3698925820
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiPageSourceProviderTest.java:
##########
@@ -193,6 +203,231 @@ public void testRemapColumnNotFound()
assertHandle(remapped.get(1), "col_x", fileSchema.getFieldCount(),
HiveType.HIVE_STRING, VARCHAR);
}
+ @Test
+ public void testRemapPredicateStaleMetastoreOrdinals()
+ {
+ // Physical Schema: the five Hudi meta columns, then [c0, c1, c2]
+ MessageType fileSchema = hudiFileSchema(3);
+
+ // A metastore synced with omit_metadata_fields=true carries no meta
columns, so "c2" is numbered 2
+ // while it physically sits at 7, and "c0" is numbered 0 while it
physically sits at 5.
+ HiveColumnHandle staleC2 = createDummyHandle("c2", 2,
HiveType.HIVE_INT, INTEGER);
+ HiveColumnHandle staleC0 = createDummyHandle("c0", 0,
HiveType.HIVE_INT, INTEGER);
+ Domain c2Domain =
Domain.create(ValueSet.ofRanges(Range.greaterThan(INTEGER, 900L)), false);
+ Domain c0Domain = Domain.singleValue(INTEGER, 7L);
+
+ TupleDomain<HiveColumnHandle> remapped =
remapPredicateColumnIndicesToPhysical(
+ fileSchema,
+ TupleDomain.withColumnDomains(Map.of(staleC2, c2Domain,
staleC0, c0Domain)),
+ false);
+
+ Map<HiveColumnHandle, Domain> domains =
remapped.getDomains().orElseThrow();
+ assertThat(domains).hasSize(2);
+ // Each domain now keys off the column's physical position, so it is
matched against that column's statistics
+ assertThat(handleOf(domains,
"c2").getBaseHiveColumnIndex()).isEqualTo(7);
+ assertThat(domains.get(handleOf(domains, "c2"))).isEqualTo(c2Domain);
+ assertThat(handleOf(domains,
"c0").getBaseHiveColumnIndex()).isEqualTo(5);
+ assertThat(domains.get(handleOf(domains, "c0"))).isEqualTo(c0Domain);
+ }
+
+ @Test
+ public void testRemapPredicateDropsColumnAbsentFromFile()
+ {
+ // Physical Schema: the five Hudi meta columns, then [c0]
+ MessageType fileSchema = hudiFileSchema(1);
+
+ HiveColumnHandle present = createDummyHandle("c0", 0,
HiveType.HIVE_INT, INTEGER);
+ // Added after this base file was written, so the file does not carry
it
+ HiveColumnHandle absent = createDummyHandle("c1", 1,
HiveType.HIVE_INT, INTEGER);
+ Domain presentDomain = Domain.singleValue(INTEGER, 1L);
+
+ TupleDomain<HiveColumnHandle> remapped =
remapPredicateColumnIndicesToPhysical(
+ fileSchema,
+ TupleDomain.withColumnDomains(Map.of(present, presentDomain,
absent, Domain.singleValue(INTEGER, 2L))),
+ false);
+
+ // The absent column is dropped rather than mapped to the projection
remap's out-of-range sentinel
+ Map<HiveColumnHandle, Domain> domains =
remapped.getDomains().orElseThrow();
+ assertThat(domains).hasSize(1);
+ assertThat(handleOf(domains,
"c0").getBaseHiveColumnIndex()).isEqualTo(5);
+ assertThat(domains.get(handleOf(domains,
"c0"))).isEqualTo(presentDomain);
+ }
+
+ @Test
+ public void testRemapPredicateWithSeveralAbsentColumnsDoesNotCollide()
+ {
+ // Physical Schema: the five Hudi meta columns, then [c0]
+ MessageType fileSchema = hudiFileSchema(1);
+
+ HiveColumnHandle firstAbsent = createDummyHandle("c1", 1,
HiveType.HIVE_INT, INTEGER);
+ HiveColumnHandle secondAbsent = createDummyHandle("c2", 2,
HiveType.HIVE_INT, INTEGER);
+
+ TupleDomain<HiveColumnHandle> remapped =
remapPredicateColumnIndicesToPhysical(
+ fileSchema,
+ TupleDomain.withColumnDomains(Map.of(
+ firstAbsent, Domain.singleValue(INTEGER, 1L),
+ secondAbsent, Domain.singleValue(INTEGER, 2L))),
+ false);
+
+ // Both would share the sentinel index, which
TupleDomain.transformKeys rejects as a duplicate key.
+ // Dropping them instead leaves nothing to push down, and the engine
still applies the filter itself.
+ assertThat(remapped.isAll()).isTrue();
+ }
+
+ @Test
+ public void testRemapPredicateKeepsOneDomainPerPhysicalColumn()
+ {
+ // Physical Schema: the five Hudi meta columns, then [c0]
+ MessageType fileSchema = hudiFileSchema(1);
+
+ // Two handles whose names differ only by case resolve to the same
file field, so both land on physical
+ // index 5 while remaining unequal to each other. The connector cannot
produce this - Hive normalises
+ // column names to lower case - but pushing both down would hand
getParquetTupleDomain one
+ // ColumnDescriptor twice, which it rejects by failing the whole split.
+ HiveColumnHandle upperCase = createDummyHandle("C0", 0,
HiveType.HIVE_INT, INTEGER);
+ HiveColumnHandle lowerCase = createDummyHandle("c0", 3,
HiveType.HIVE_INT, INTEGER);
+ Domain firstDomain =
Domain.create(ValueSet.ofRanges(Range.greaterThan(INTEGER, 10L)), false);
+ // Insertion-ordered so that "first wins" is a deterministic assertion
+ Map<HiveColumnHandle, Domain> predicate = new LinkedHashMap<>();
+ predicate.put(upperCase, firstDomain);
+ predicate.put(lowerCase,
Domain.create(ValueSet.ofRanges(Range.lessThan(INTEGER, 20L)), false));
+
+ TupleDomain<HiveColumnHandle> remapped =
remapPredicateColumnIndicesToPhysical(
+ fileSchema, TupleDomain.withColumnDomains(predicate), false);
+
+ // Only the first is pushed down, and no IllegalArgumentException
escapes
+ Map<HiveColumnHandle, Domain> domains =
remapped.getDomains().orElseThrow();
+ assertThat(domains).hasSize(1);
+ HiveColumnHandle survivor = handleOf(domains, "C0");
+ assertThat(survivor.getBaseHiveColumnIndex()).isEqualTo(5);
+ assertThat(domains.get(survivor)).isEqualTo(firstDomain);
+ }
+
+ @Test
+ public void testRemapPreservesTheBaseTypeOfADereferenceHandle()
+ {
+ // Physical Schema: the five Hudi meta columns, then [c0]
+ MessageType fileSchema = hudiFileSchema(1);
+
+ // A handle projecting one field out of a struct column. HudiMetadata
does not implement applyProjection,
+ // so the connector never builds one today, but the remap has to
rebuild it without corrupting it: the
+ // constructor's type argument is the BASE column's type, while
getType() is the projected field's.
+ RowType baseType = RowType.rowType(RowType.field("f", INTEGER));
+ HiveColumnHandle dereference = new HiveColumnHandle(
+ "c0",
+ 0,
+ HiveType.valueOf("struct<f:int>"),
+ baseType,
+ Optional.of(new HiveColumnProjectionInfo(List.of(0),
List.of("f"), HiveType.HIVE_INT, INTEGER)),
+ HiveColumnHandle.ColumnType.REGULAR,
+ Optional.empty());
+
+ HiveColumnHandle remapped = remapColumnIndicesToPhysical(fileSchema,
List.of(dereference), false).get(0);
+
+ assertThat(remapped.getBaseHiveColumnIndex())
+ .as("physical index")
+ .isEqualTo(5);
+ assertThat(remapped.getBaseType())
+ .as("base type, which is what the parquet page source reads")
+ .isEqualTo(baseType);
+ assertThat(remapped.getType())
+ .as("projected field type")
+ .isEqualTo(INTEGER);
+ assertThat(remapped.getHiveColumnProjectionInfo())
+ .as("projection info")
+ .isEqualTo(dereference.getHiveColumnProjectionInfo());
+ }
+
+ @Test
+ public void testRemapPredicateAllAndNonePassThrough()
+ {
+ MessageType fileSchema = hudiFileSchema(1);
+
+ assertThat(remapPredicateColumnIndicesToPhysical(fileSchema,
TupleDomain.<HiveColumnHandle>all(), false))
+ .isEqualTo(TupleDomain.all());
+ assertThat(remapPredicateColumnIndicesToPhysical(fileSchema,
TupleDomain.<HiveColumnHandle>none(), false))
+ .isEqualTo(TupleDomain.none());
+ }
+
+ @Test
+ public void testRemapPredicateCaseSensitivity()
+ {
+ // Physical Schema: the five Hudi meta columns, then [c0]
+ MessageType fileSchema = hudiFileSchema(1);
+
+ HiveColumnHandle upperCase = createDummyHandle("C0", 0,
HiveType.HIVE_INT, INTEGER);
+ TupleDomain<HiveColumnHandle> predicate =
TupleDomain.withColumnDomains(Map.of(upperCase, Domain.singleValue(INTEGER,
1L)));
+
+ // Case-insensitive: "C0" resolves to the file's "c0" at physical
index 5
+ Map<HiveColumnHandle, Domain> insensitive =
remapPredicateColumnIndicesToPhysical(fileSchema, predicate, false)
+ .getDomains().orElseThrow();
+ assertThat(handleOf(insensitive,
"C0").getBaseHiveColumnIndex()).isEqualTo(5);
+
+ // Case-sensitive: no match, so the domain is dropped instead of being
left on a stale ordinal
+ assertThat(remapPredicateColumnIndicesToPhysical(fileSchema,
predicate, true).isAll()).isTrue();
+ }
+
+ @Test
+ public void testRemapPredicatePreservesEveryOtherHandleAttribute()
+ {
+ // Physical Schema: the five Hudi meta columns, then [c0]
+ MessageType fileSchema = hudiFileSchema(1);
+
+ HiveColumnHandle original = new HiveColumnHandle(
+ "c0",
+ 0,
+ HiveType.HIVE_INT,
+ INTEGER,
+ Optional.empty(),
+ HiveColumnHandle.ColumnType.REGULAR,
+ Optional.of("a comment"));
+ Domain domain =
Domain.create(ValueSet.ofRanges(Range.greaterThan(INTEGER, 900L)), true);
+
+ Map<HiveColumnHandle, Domain> domains =
remapPredicateColumnIndicesToPhysical(
+ fileSchema,
+ TupleDomain.withColumnDomains(Map.of(original, domain)),
+ false)
+ .getDomains().orElseThrow();
+
+ HiveColumnHandle remapped = handleOf(domains, "c0");
+ assertHandle(remapped, "c0", 5, HiveType.HIVE_INT, INTEGER);
+ assertThat(remapped.getComment())
+ .as("Comment mismatch for c0")
+ .isEqualTo(Optional.of("a comment"));
+ assertThat(domains.get(remapped))
+ .as("Domain mismatch for c0")
+ .isEqualTo(domain);
+ }
+
+ /**
+ * Builds a file schema laid out like a Hudi base file: the five {@code
_hoodie_*} meta columns followed by
+ * {@code dataColumnCount} int columns named {@code c0..cN}. A metastore
synced with
+ * {@code hoodie.datasource.hive_sync.omit_metadata_fields=true} omits the
meta columns, so a data column's
+ * metastore ordinal is its physical ordinal minus five.
+ */
+ private static MessageType hudiFileSchema(int dataColumnCount)
+ {
+ List<org.apache.parquet.schema.Type> fields = new ArrayList<>();
+ for (String metaColumn : List.of("_hoodie_commit_time",
"_hoodie_commit_seqno", "_hoodie_record_key", "_hoodie_partition_path",
"_hoodie_file_name")) {
Review Comment:
Done 40261aca65ce
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiPredicatePushdownColumnOrdinals.java:
##########
@@ -0,0 +1,354 @@
+/*
+ * 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.local.LocalInputFile;
+import io.trino.parquet.ParquetReaderOptions;
+import io.trino.plugin.base.metrics.FileFormatDataSourceStats;
+import io.trino.plugin.hive.HiveColumnHandle;
+import io.trino.plugin.hive.parquet.ParquetReaderConfig;
+import io.trino.plugin.hudi.file.HudiBaseFile;
+import io.trino.spi.SplitWeight;
+import io.trino.spi.connector.ColumnHandle;
+import io.trino.spi.connector.ConnectorPageSource;
+import io.trino.spi.connector.ConnectorSession;
+import io.trino.spi.connector.DynamicFilter;
+import io.trino.spi.predicate.Domain;
+import io.trino.spi.predicate.Range;
+import io.trino.spi.predicate.TupleDomain;
+import io.trino.spi.predicate.ValueSet;
+import io.trino.spi.type.Type;
+import io.trino.testing.MaterializedResult;
+import io.trino.testing.TestingConnectorSession;
+import org.apache.parquet.conf.PlainParquetConfiguration;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.SimpleGroupFactory;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.io.LocalOutputFile;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.Types;
+import org.joda.time.DateTimeZone;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.OptionalLong;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+import static io.trino.metastore.HiveType.HIVE_INT;
+import static io.trino.plugin.hive.HiveColumnHandle.ColumnType.REGULAR;
+import static io.trino.plugin.hive.HiveColumnHandle.createBaseColumn;
+import static io.trino.plugin.hudi.HudiPageSourceProvider.createPageSource;
+import static io.trino.spi.type.IntegerType.INTEGER;
+import static io.trino.testing.MaterializedResult.materializeSourceDataStream;
+import static java.lang.Integer.parseInt;
+import static org.apache.parquet.schema.Type.Repetition.OPTIONAL;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Reads a base file whose physical column order does not match the
metastore's, the layout hive sync produces
+ * with {@code hoodie.datasource.hive_sync.omit_metadata_fields=true}: the
five {@code _hoodie_*} meta columns are
+ * absent from the metastore, so every data column's metastore ordinal is five
below its physical position.
+ * <p>
+ * With {@code hudi.parquet.use-column-names=false} the parquet page source
resolves columns positionally, so a
+ * predicate whose handle still carries the metastore ordinal lands on
whichever column physically sits there and
+ * row groups get pruned on that column's statistics. The fixture makes that
observable: {@code c7} grows with the
+ * row index while every other data column stays in 0..9, so a domain meant
for {@code c7} but applied to any other
+ * column excludes every row group and the read returns nothing.
+ * <p>
+ * Note that the shadowed column has to be part of the PROJECTION for the
damage to appear: {@code
+ * descriptorsByPath} is derived from the projection, so a domain resolving to
a column the query does not read
+ * finds no descriptor and is discarded instead. Do not "simplify" the
projections below to the predicate column
+ * alone - that turns these tests green against the unfixed code.
+ */
+class TestHudiPredicatePushdownColumnOrdinals
+{
+ private static final List<String> META_COLUMNS = List.of(
+ "_hoodie_commit_time",
+ "_hoodie_commit_seqno",
+ "_hoodie_record_key",
+ "_hoodie_partition_path",
+ "_hoodie_file_name");
+ private static final int DATA_COLUMN_COUNT = 10;
+ /** The column the predicate is on: physically at 12, but numbered 7 by a
metastore without the meta columns. */
+ private static final String PREDICATE_COLUMN = "c7";
+ /** The column physically sitting at {@code c7}'s stale ordinal, and
therefore the one that shadows it. */
+ private static final String SHADOWED_COLUMN = "c2";
+ private static final int ROW_COUNT = 1000;
+ private static final long THRESHOLD = 900;
+ private static final int MATCHING_ROW_COUNT = (int) (ROW_COUNT - THRESHOLD
- 1);
+
+ @TempDir
+ static Path tempDir;
+
+ private static Path baseFile;
+
+ @BeforeAll
+ static void writeBaseFile()
+ throws IOException
+ {
+ MessageType schema = fileSchema();
+ baseFile = tempDir.resolve("base_file.parquet");
+ SimpleGroupFactory groupFactory = new SimpleGroupFactory(schema);
+ try (ParquetWriter<Group> writer = ExampleParquetWriter.builder(new
LocalOutputFile(baseFile))
+ .withType(schema)
+ .withConf(new PlainParquetConfiguration())
+ .withRowGroupSize(1024L)
+ .withPageSize(512)
+ .build()) {
+ for (int row = 0; row < ROW_COUNT; row++) {
+ Group group = groupFactory.newGroup();
+ for (String metaColumn : META_COLUMNS) {
+ group.append(metaColumn, metaColumn + "_" + row);
+ }
+ for (int column = 0; column < DATA_COLUMN_COUNT; column++) {
+ String columnName = "c" + column;
+ group.append(columnName,
columnName.equals(PREDICATE_COLUMN) ? row : row % 10);
+ }
+ writer.write(group);
+ }
+ }
+ // The writer flushes a row group whenever the buffered size is over
withRowGroupSize, checked every
+ // parquet.page.size.row.check.min records (100 by default), which is
what actually splits this file.
+ // Assert the outcome rather than the knobs: with a single row group
there would be nothing to prune,
+ // and every test below would pass without proving anything.
+ assertThat(rowGroupCount(baseFile)).as("row groups
written").isGreaterThan(1);
+ }
+
+ @Test
+ public void testPredicateOnStaleOrdinalKeepsMatchingRows()
+ throws Exception
+ {
+ List<HiveColumnHandle> projection =
List.of(dataColumn(SHADOWED_COLUMN), dataColumn(PREDICATE_COLUMN));
+
+ MaterializedResult result = read(projection,
greaterThanThreshold(PREDICATE_COLUMN), false, DynamicFilter.EMPTY);
+
+ // The shadowed column never leaves 0..9, so a domain of "> 900"
applied to it prunes every row group
+ assertThat(matchingRowCount(result, projection, PREDICATE_COLUMN))
+ .as("rows matching %s > %s", PREDICATE_COLUMN, THRESHOLD)
+ .isEqualTo(MATCHING_ROW_COUNT);
+ }
+
+ @Test
+ public void testPredicateOnStaleOrdinalStillPrunesRowGroups()
+ throws Exception
+ {
+ List<HiveColumnHandle> projection =
List.of(dataColumn(SHADOWED_COLUMN), dataColumn(PREDICATE_COLUMN));
+
+ MaterializedResult result = read(projection,
greaterThanThreshold(PREDICATE_COLUMN), false, DynamicFilter.EMPTY);
+
+ // Correct results alone would also be produced by pushing nothing
down; reading fewer rows than the file
+ // holds is only possible if the domain reached the column it was
written for, and the matching rows must
+ // survive that pruning
+ assertThat(result.getRowCount())
+ .as("rows read out of %s", ROW_COUNT)
+ .isLessThan(ROW_COUNT);
+ assertThat(matchingRowCount(result, projection, PREDICATE_COLUMN))
+ .as("rows matching %s > %s after pruning", PREDICATE_COLUMN,
THRESHOLD)
+ .isEqualTo(MATCHING_ROW_COUNT);
+ }
+
+ @Test
+ public void testStaleOrdinalArrivingThroughADynamicFilter()
+ throws Exception
+ {
+ List<HiveColumnHandle> projection =
List.of(dataColumn(SHADOWED_COLUMN), dataColumn(PREDICATE_COLUMN));
+
+ // A dynamic filter reaches getCombinedPredicate by its own route, and
its handles carry the same stale
+ // metastore ordinals the split's predicate does
+ MaterializedResult result = read(projection, TupleDomain.all(), false,
+ dynamicFilterOn(greaterThanThreshold(PREDICATE_COLUMN)));
+
+ assertThat(matchingRowCount(result, projection, PREDICATE_COLUMN))
+ .as("rows matching a dynamic filter of %s > %s",
PREDICATE_COLUMN, THRESHOLD)
+ .isEqualTo(MATCHING_ROW_COUNT);
+ }
+
+ @Test
+ public void testPredicateOnColumnAddedAfterBaseFileWasWritten()
+ throws Exception
+ {
+ // The metastore carries one column more than this base file does,
numbered 10 - an ordinal that is still
+ // in range physically, where it picks out "c5"
+ String addedColumn = "c" + DATA_COLUMN_COUNT;
+ List<HiveColumnHandle> projection = List.of(dataColumn("c5"),
dataColumn(PREDICATE_COLUMN), dataColumn(addedColumn));
+
+ // IS NULL, not a range: the added column is null in every row of this
base file, so this predicate is
+ // satisfied by all of them. A range predicate would be unsatisfiable
here and the buggy read's empty
+ // result would be the right answer by accident.
+ MaterializedResult result = read(projection,
+ TupleDomain.withColumnDomains(Map.of(dataColumn(addedColumn),
Domain.onlyNull(INTEGER))),
+ false, DynamicFilter.EMPTY);
+
+ // A column the file does not carry has to be dropped from the
pushed-down predicate. Pushed positionally
+ // it would land on "c5", which has no nulls at all, and every row
group would be pruned.
+ assertThat(result.getRowCount()).as("rows read").isEqualTo(ROW_COUNT);
+
assertThat(result.getMaterializedRows().getFirst().getField(2)).as("value of
%s", addedColumn).isNull();
+ }
+
+ @Test
+ public void testPositionalAndNameBasedResolutionAgree()
+ throws Exception
+ {
+ List<HiveColumnHandle> projection =
List.of(dataColumn(SHADOWED_COLUMN), dataColumn(PREDICATE_COLUMN));
+ TupleDomain<HiveColumnHandle> predicate =
greaterThanThreshold(PREDICATE_COLUMN);
+
+ MaterializedResult positional = read(projection, predicate, false,
DynamicFilter.EMPTY);
+ MaterializedResult byName = read(projection, predicate, true,
DynamicFilter.EMPTY);
+
+ // Anchor the comparison: both modes regressing to no pushdown at all
would otherwise agree happily
+ assertThat(byName.getRowCount()).as("rows read with
use-column-names=true").isLessThan(ROW_COUNT);
+ assertThat(positional.getMaterializedRows())
+ .as("hudi.parquet.use-column-names=false must read what
use-column-names=true reads")
+ .isEqualTo(byName.getMaterializedRows());
+ }
+
+ /**
+ * Reads the whole base file through the page source the connector builds
for a split with no log files, which
+ * is the only path on which it enables predicate pushdown.
+ */
+ private static MaterializedResult read(
+ List<HiveColumnHandle> projection,
+ TupleDomain<HiveColumnHandle> predicate,
+ boolean useParquetColumnNames,
+ DynamicFilter dynamicFilter)
+ throws Exception
+ {
+ long fileSize = Files.size(baseFile);
+ HudiSplit split = new HudiSplit(
+ new HudiBaseFile(baseFile.toString(),
baseFile.getFileName().toString(), fileSize, 0, 0, fileSize),
+ List.of(),
+ "000",
+ predicate,
+ List.of(),
+ SplitWeight.standard());
+ HudiSessionProperties sessionProperties = new HudiSessionProperties(
+ new
HudiConfig().setUseParquetColumnNames(useParquetColumnNames),
+ new ParquetReaderConfig());
+ ConnectorSession session = TestingConnectorSession.builder()
+ .setPropertyMetadata(sessionProperties.getSessionProperties())
+ .build();
+
+ List<Type> types =
projection.stream().map(HiveColumnHandle::getType).toList();
+ try (ConnectorPageSource pageSource = createPageSource(
+ session,
+ projection,
+ split,
+ new LocalInputFile(baseFile.toFile()),
+ baseFile.toString(),
+ 0L,
+ fileSize,
+ OptionalLong.of(fileSize),
+ new FileFormatDataSourceStats(),
+ ParquetReaderOptions.builder().build(),
+ DateTimeZone.UTC,
+ dynamicFilter,
+ true)) {
+ return materializeSourceDataStream(session, pageSource,
types).toTestTypes();
+ }
+ }
+
+ private static MessageType fileSchema()
Review Comment:
Done 40261aca65ce
--
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]
