FrankChen021 commented on code in PR #20729: URL: https://github.com/apache/druid/pull/20729#discussion_r4412536463
########## extensions-contrib/tsfile-extensions/src/main/java/org/apache/druid/data/input/tsfile/TsFileInputEntityReader.java: ########## @@ -0,0 +1,371 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.druid.data.input.tsfile; + +import org.apache.druid.data.input.InputEntity; +import org.apache.druid.data.input.InputRow; +import org.apache.druid.data.input.InputRowSchema; +import org.apache.druid.data.input.IntermediateRowParsingReader; +import org.apache.druid.data.input.impl.MapInputRowParser; +import org.apache.druid.error.InvalidInput; +import org.apache.druid.java.util.common.CloseableIterators; +import org.apache.druid.java.util.common.parsers.CloseableIterator; +import org.apache.druid.java.util.common.parsers.ParseException; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.exception.read.ReadProcessException; +import org.apache.tsfile.exception.write.NoMeasurementException; +import org.apache.tsfile.exception.write.NoTableException; +import org.apache.tsfile.file.metadata.TableSchema; +import org.apache.tsfile.read.query.dataset.ResultSet; +import org.apache.tsfile.read.v4.ITsFileReader; +import org.apache.tsfile.read.v4.TsFileReaderBuilder; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.joda.time.Interval; + +import javax.annotation.Nullable; + +import java.io.File; +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.ArrayList; +import java.util.Base64; +import java.util.Collections; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.NoSuchElementException; +import java.util.Set; +import java.util.stream.Collectors; + +/** + * Reads the rows of a TsFile file with the TsFile table-model reader and converts them into + * {@link MapBasedInputRow}-compatible maps. + * <p> + * TsFile stores its own timestamp, which is exposed to Druid under the name {@value TsFileInputFormat#TIME_COLUMN_NAME}. + * Tag, field, and attribute columns are exposed as-is. + */ +public class TsFileInputEntityReader extends IntermediateRowParsingReader<Map<String, Object>> +{ + private final InputRowSchema inputRowSchema; + private final InputEntity source; + @Nullable + private final File temporaryDirectory; + @Nullable + private final List<String> tables; + @Nullable + private final List<String> columns; + @Nullable + private final Interval interval; + + TsFileInputEntityReader( + InputRowSchema inputRowSchema, + InputEntity source, + @Nullable File temporaryDirectory, + @Nullable List<String> tables, + @Nullable List<String> columns, + @Nullable Interval interval + ) + { + this.inputRowSchema = inputRowSchema; + this.source = source; + this.temporaryDirectory = temporaryDirectory; + this.tables = tables; + this.columns = columns; + this.interval = interval; + } + + @Override + protected CloseableIterator<Map<String, Object>> intermediateRowIterator() throws IOException + { + final InputEntity.CleanableFile materialized = + source.fetch(temporaryDirectory, new byte[InputEntity.DEFAULT_FETCH_BUFFER_SIZE]); + final ITsFileReader reader; + try { + reader = new TsFileReaderBuilder().file(materialized.file()).build(); + } + catch (IOException | RuntimeException e) { + materialized.close(); + throw e; + } + + final List<String> tableNames; + try { + tableNames = resolveTableNames(reader); + } + catch (IOException | RuntimeException e) { + reader.close(); + materialized.close(); + throw e; + } + + return CloseableIterators.wrap( + new RowIterator(reader, tableNames), + () -> { + reader.close(); + materialized.close(); + } + ); + } + + @Override + protected InputEntity source() + { + return source; + } + + @Override + protected List<InputRow> parseInputRows(Map<String, Object> intermediateRow) throws ParseException + { + return Collections.singletonList(MapInputRowParser.parse(inputRowSchema, intermediateRow)); + } + + @Override + protected List<Map<String, Object>> toMap(Map<String, Object> intermediateRow) + { + return Collections.singletonList(intermediateRow); + } + + private List<String> resolveTableNames(ITsFileReader reader) throws IOException + { + final List<TableSchema> allTables = reader.getAllTableSchema(); + if (allTables.isEmpty()) { + throw InvalidInput.exception( + "TsFile[%s] contains no table schema. Only table-model TsFiles are supported; tree-model files " + + "are not.", + source.getUri() + ); + } + if (tables == null || tables.isEmpty()) { + return allTables.stream().map(TableSchema::getTableName).collect(Collectors.toList()); + } + + final Set<String> available = + allTables.stream().map(schema -> schema.getTableName().toLowerCase(Locale.ROOT)).collect(Collectors.toSet()); + for (String table : tables) { + if (!available.contains(table.toLowerCase(Locale.ROOT))) { + throw InvalidInput.exception( + "Table[%s] does not exist in the file. Available tables: %s", + table, + allTables.stream().map(TableSchema::getTableName).collect(Collectors.toList()) + ); + } + } + return tables; + } + + private List<String> selectColumns(TableSchema tableSchema) + { + if (columns != null && !columns.isEmpty()) { Review Comment: The license-registration finding is resolved. The projection change is not fully safe yet: `selected.size() == physicalColumns.size()` also occurs for normal fixed-schema batch ingestion, where rebuilding from dimension names and metric output names can drop transform or aggregator input columns; the tag-only/empty fallback can also expand an EXTERN projection to every physical column and fail on an omitted unsupported field. Please preserve the ColumnsFilter-selected inputs for batch ingestion and avoid falling back to all physical columns. Reviewed 18 of 18 changed files. <!-- mergelens:review --> ########## extensions-contrib/tsfile-extensions/src/main/java/org/apache/druid/data/input/tsfile/TsFileInputEntityReader.java: ########## @@ -0,0 +1,411 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.druid.data.input.tsfile; + +import org.apache.druid.data.input.ColumnsFilter; +import org.apache.druid.data.input.InputEntity; +import org.apache.druid.data.input.InputRow; +import org.apache.druid.data.input.InputRowSchema; +import org.apache.druid.data.input.IntermediateRowParsingReader; +import org.apache.druid.data.input.impl.DimensionsSpec; +import org.apache.druid.data.input.impl.MapInputRowParser; +import org.apache.druid.error.InvalidInput; +import org.apache.druid.java.util.common.CloseableIterators; +import org.apache.druid.java.util.common.parsers.CloseableIterator; +import org.apache.druid.java.util.common.parsers.ParseException; +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.exception.read.ReadProcessException; +import org.apache.tsfile.exception.write.NoMeasurementException; +import org.apache.tsfile.exception.write.NoTableException; +import org.apache.tsfile.file.metadata.TableSchema; +import org.apache.tsfile.read.query.dataset.ResultSet; +import org.apache.tsfile.read.v4.ITsFileReader; +import org.apache.tsfile.read.v4.TsFileReaderBuilder; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.joda.time.Interval; + +import javax.annotation.Nullable; + +import java.io.File; +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.ArrayList; +import java.util.Base64; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.NoSuchElementException; +import java.util.Set; +import java.util.stream.Collectors; + +/** + * Reads the rows of a TsFile file with the TsFile table-model reader and converts them into + * {@link MapBasedInputRow}-compatible maps. + * <p> + * TsFile stores its own timestamp, which is exposed to Druid under the name {@value TsFileInputFormat#TIME_COLUMN_NAME}. + * Tag, field, and attribute columns are exposed as-is. + */ +public class TsFileInputEntityReader extends IntermediateRowParsingReader<Map<String, Object>> +{ + private final InputRowSchema inputRowSchema; + private final InputEntity source; + @Nullable + private final File temporaryDirectory; + @Nullable + private final List<String> tables; + @Nullable + private final List<String> columns; + @Nullable + private final Interval interval; + + TsFileInputEntityReader( + InputRowSchema inputRowSchema, + InputEntity source, + @Nullable File temporaryDirectory, + @Nullable List<String> tables, + @Nullable List<String> columns, + @Nullable Interval interval + ) + { + this.inputRowSchema = inputRowSchema; + this.source = source; + this.temporaryDirectory = temporaryDirectory; + this.tables = tables; + this.columns = columns; + this.interval = interval; + } + + @Override + protected CloseableIterator<Map<String, Object>> intermediateRowIterator() throws IOException + { + final InputEntity.CleanableFile materialized = + source.fetch(temporaryDirectory, new byte[InputEntity.DEFAULT_FETCH_BUFFER_SIZE]); + final ITsFileReader reader; + try { + reader = new TsFileReaderBuilder().file(materialized.file()).build(); + } + catch (IOException | RuntimeException e) { + materialized.close(); + throw e; + } + + final List<String> tableNames; + try { + tableNames = resolveTableNames(reader); + } + catch (IOException | RuntimeException e) { + reader.close(); + materialized.close(); + throw e; + } + + return CloseableIterators.wrap( + new RowIterator(reader, tableNames), + () -> { + reader.close(); + materialized.close(); + } + ); + } + + @Override + protected InputEntity source() + { + return source; + } + + @Override + protected List<InputRow> parseInputRows(Map<String, Object> intermediateRow) throws ParseException + { + return Collections.singletonList(MapInputRowParser.parse(inputRowSchema, intermediateRow)); + } + + @Override + protected List<Map<String, Object>> toMap(Map<String, Object> intermediateRow) + { + return Collections.singletonList(intermediateRow); + } + + private List<String> resolveTableNames(ITsFileReader reader) throws IOException + { + final List<TableSchema> allTables = reader.getAllTableSchema(); + if (allTables.isEmpty()) { + throw InvalidInput.exception( + "TsFile[%s] contains no table schema. Only table-model TsFiles are supported; tree-model files " + + "are not.", + source.getUri() + ); + } + if (tables == null || tables.isEmpty()) { + return allTables.stream().map(TableSchema::getTableName).collect(Collectors.toList()); + } + + final Set<String> available = + allTables.stream().map(schema -> schema.getTableName().toLowerCase(Locale.ROOT)).collect(Collectors.toSet()); + for (String table : tables) { + if (!available.contains(table.toLowerCase(Locale.ROOT))) { + throw InvalidInput.exception( + "Table[%s] does not exist in the file. Available tables: %s", + table, + allTables.stream().map(TableSchema::getTableName).collect(Collectors.toList()) + ); + } + } + return tables; + } + + private List<String> selectColumns(TableSchema tableSchema) + { + if (columns != null && !columns.isEmpty()) { + return columns; + } + + final List<IMeasurementSchema> columnSchemas = tableSchema.getColumnSchemas(); + final List<ColumnCategory> columnCategories = tableSchema.getColumnTypes(); + final List<String> physicalColumns = new ArrayList<>(); + final Set<String> tagColumns = new HashSet<>(); + for (int i = 0; i < columnSchemas.size(); i++) { + final String columnName = columnSchemas.get(i).getMeasurementName(); + if (TsFileInputFormat.TIME_COLUMN_NAME.equalsIgnoreCase(columnName)) { + continue; + } + physicalColumns.add(columnName); + if (columnCategories.get(i) == ColumnCategory.TAG) { + tagColumns.add(columnName); + } + } + + final ColumnsFilter columnsFilter = inputRowSchema.getColumnsFilter(); + List<String> selected = columnsFilter == null + ? physicalColumns + : physicalColumns.stream().filter(columnsFilter::apply).collect(Collectors.toList()); + + // EXTERN passes ColumnsFilter.all(), so the projection has to come from the schema instead: its fixed + // dimensions are the external table's column signature. + if (selected.size() == physicalColumns.size()) { Review Comment: [P1] Preserve required batch input columns **Finding:** This condition is not specific to EXTERN: a normal fixed-schema batch InputRowSchema can have a ColumnsFilter that selects every physical column. The code then rebuilds the projection from only dimension names and metric output names, dropping transform-required columns or aggregator requiredFields whose names differ from their output metrics; those transforms or aggregators receive nulls and ingested values can be silently wrong. **Suggestion:** Apply the fixed-dimension signature projection only to the EXTERN/all-columns case, or retain the ColumnsFilter result for normal batch ingestion and add coverage for an aliased metric or transform input. ########## extensions-contrib/tsfile-extensions/src/main/java/org/apache/druid/data/input/tsfile/TsFileInputEntityReader.java: ########## @@ -0,0 +1,411 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.druid.data.input.tsfile; + +import org.apache.druid.data.input.ColumnsFilter; +import org.apache.druid.data.input.InputEntity; +import org.apache.druid.data.input.InputRow; +import org.apache.druid.data.input.InputRowSchema; +import org.apache.druid.data.input.IntermediateRowParsingReader; +import org.apache.druid.data.input.impl.DimensionsSpec; +import org.apache.druid.data.input.impl.MapInputRowParser; +import org.apache.druid.error.InvalidInput; +import org.apache.druid.java.util.common.CloseableIterators; +import org.apache.druid.java.util.common.parsers.CloseableIterator; +import org.apache.druid.java.util.common.parsers.ParseException; +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.exception.read.ReadProcessException; +import org.apache.tsfile.exception.write.NoMeasurementException; +import org.apache.tsfile.exception.write.NoTableException; +import org.apache.tsfile.file.metadata.TableSchema; +import org.apache.tsfile.read.query.dataset.ResultSet; +import org.apache.tsfile.read.v4.ITsFileReader; +import org.apache.tsfile.read.v4.TsFileReaderBuilder; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.joda.time.Interval; + +import javax.annotation.Nullable; + +import java.io.File; +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.ArrayList; +import java.util.Base64; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.NoSuchElementException; +import java.util.Set; +import java.util.stream.Collectors; + +/** + * Reads the rows of a TsFile file with the TsFile table-model reader and converts them into + * {@link MapBasedInputRow}-compatible maps. + * <p> + * TsFile stores its own timestamp, which is exposed to Druid under the name {@value TsFileInputFormat#TIME_COLUMN_NAME}. + * Tag, field, and attribute columns are exposed as-is. + */ +public class TsFileInputEntityReader extends IntermediateRowParsingReader<Map<String, Object>> +{ + private final InputRowSchema inputRowSchema; + private final InputEntity source; + @Nullable + private final File temporaryDirectory; + @Nullable + private final List<String> tables; + @Nullable + private final List<String> columns; + @Nullable + private final Interval interval; + + TsFileInputEntityReader( + InputRowSchema inputRowSchema, + InputEntity source, + @Nullable File temporaryDirectory, + @Nullable List<String> tables, + @Nullable List<String> columns, + @Nullable Interval interval + ) + { + this.inputRowSchema = inputRowSchema; + this.source = source; + this.temporaryDirectory = temporaryDirectory; + this.tables = tables; + this.columns = columns; + this.interval = interval; + } + + @Override + protected CloseableIterator<Map<String, Object>> intermediateRowIterator() throws IOException + { + final InputEntity.CleanableFile materialized = + source.fetch(temporaryDirectory, new byte[InputEntity.DEFAULT_FETCH_BUFFER_SIZE]); + final ITsFileReader reader; + try { + reader = new TsFileReaderBuilder().file(materialized.file()).build(); + } + catch (IOException | RuntimeException e) { + materialized.close(); + throw e; + } + + final List<String> tableNames; + try { + tableNames = resolveTableNames(reader); + } + catch (IOException | RuntimeException e) { + reader.close(); + materialized.close(); + throw e; + } + + return CloseableIterators.wrap( + new RowIterator(reader, tableNames), + () -> { + reader.close(); + materialized.close(); + } + ); + } + + @Override + protected InputEntity source() + { + return source; + } + + @Override + protected List<InputRow> parseInputRows(Map<String, Object> intermediateRow) throws ParseException + { + return Collections.singletonList(MapInputRowParser.parse(inputRowSchema, intermediateRow)); + } + + @Override + protected List<Map<String, Object>> toMap(Map<String, Object> intermediateRow) + { + return Collections.singletonList(intermediateRow); + } + + private List<String> resolveTableNames(ITsFileReader reader) throws IOException + { + final List<TableSchema> allTables = reader.getAllTableSchema(); + if (allTables.isEmpty()) { + throw InvalidInput.exception( + "TsFile[%s] contains no table schema. Only table-model TsFiles are supported; tree-model files " + + "are not.", + source.getUri() + ); + } + if (tables == null || tables.isEmpty()) { + return allTables.stream().map(TableSchema::getTableName).collect(Collectors.toList()); + } + + final Set<String> available = + allTables.stream().map(schema -> schema.getTableName().toLowerCase(Locale.ROOT)).collect(Collectors.toSet()); + for (String table : tables) { + if (!available.contains(table.toLowerCase(Locale.ROOT))) { + throw InvalidInput.exception( + "Table[%s] does not exist in the file. Available tables: %s", + table, + allTables.stream().map(TableSchema::getTableName).collect(Collectors.toList()) + ); + } + } + return tables; + } + + private List<String> selectColumns(TableSchema tableSchema) + { + if (columns != null && !columns.isEmpty()) { + return columns; + } + + final List<IMeasurementSchema> columnSchemas = tableSchema.getColumnSchemas(); + final List<ColumnCategory> columnCategories = tableSchema.getColumnTypes(); + final List<String> physicalColumns = new ArrayList<>(); + final Set<String> tagColumns = new HashSet<>(); + for (int i = 0; i < columnSchemas.size(); i++) { + final String columnName = columnSchemas.get(i).getMeasurementName(); + if (TsFileInputFormat.TIME_COLUMN_NAME.equalsIgnoreCase(columnName)) { + continue; + } + physicalColumns.add(columnName); + if (columnCategories.get(i) == ColumnCategory.TAG) { + tagColumns.add(columnName); + } + } + + final ColumnsFilter columnsFilter = inputRowSchema.getColumnsFilter(); + List<String> selected = columnsFilter == null + ? physicalColumns + : physicalColumns.stream().filter(columnsFilter::apply).collect(Collectors.toList()); + + // EXTERN passes ColumnsFilter.all(), so the projection has to come from the schema instead: its fixed + // dimensions are the external table's column signature. + if (selected.size() == physicalColumns.size()) { + final DimensionsSpec dimensionsSpec = inputRowSchema.getDimensionsSpec(); + if (dimensionsSpec.hasFixedDimensions()) { + final Set<String> wanted = new HashSet<>(dimensionsSpec.getDimensionNames()); + if (inputRowSchema.getMetricNames() != null) { + wanted.addAll(inputRowSchema.getMetricNames()); + } + final List<String> projected = physicalColumns.stream().filter(wanted::contains).collect(Collectors.toList()); + if (!projected.isEmpty()) { + selected = projected; + } + } + } + + // The TsFile table reader only produces blocks while it has a measurement column to read, so a projection + // made up of tag columns alone would yield no rows at all. Keep every column in that case. + if (!selected.isEmpty() && tagColumns.containsAll(selected)) { Review Comment: [P2] Do not broaden tag-only projections **Finding:** When an EXTERN signature selects only tags or no physical column matches the schema, this fallback selects every physical column. An omitted VECTOR, OBJECT, ROW, or UNKNOWN column is then passed to readValue and can fail the task even though it was not requested, leaving the prior projection issue unresolved for these valid edge cases. **Suggestion:** Preserve the requested projection and add a supported measurement only as a separate query carrier when TsFile requires one; otherwise handle the no-carrier case explicitly instead of reading every physical column. -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
