github-advanced-security[bot] commented on code in PR #19379: URL: https://github.com/apache/druid/pull/19379#discussion_r4067178912
########## processing/src/main/java/org/apache/druid/segment/transform/ScanTransformer.java: ########## @@ -0,0 +1,337 @@ +/* + * 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.segment.transform; + +import org.apache.druid.data.input.InputRow; +import org.apache.druid.data.input.InputRowListPlusRawValues; +import org.apache.druid.data.input.ListBasedInputRow; +import org.apache.druid.data.input.MapBasedInputRow; +import org.apache.druid.error.DruidException; +import org.apache.druid.java.util.common.ISE; +import org.apache.druid.java.util.common.Intervals; +import org.apache.druid.query.DataSource; +import org.apache.druid.query.QueryContexts; +import org.apache.druid.query.UnnestDataSource; +import org.apache.druid.query.scan.ScanQuery; +import org.apache.druid.segment.ColumnSelectorFactory; +import org.apache.druid.segment.ColumnValueSelector; +import org.apache.druid.segment.Cursor; +import org.apache.druid.segment.CursorBuildSpec; +import org.apache.druid.segment.CursorFactory; +import org.apache.druid.segment.CursorHolder; +import org.apache.druid.segment.Segment; +import org.apache.druid.segment.SegmentMapFunction; +import org.apache.druid.segment.VirtualColumn; +import org.apache.druid.segment.column.ColumnHolder; +import org.apache.druid.segment.column.ColumnType; +import org.apache.druid.segment.column.RowSignature; +import org.apache.druid.segment.filter.Filters; +import org.apache.druid.timeline.SegmentId; +import org.joda.time.Interval; + +import javax.annotation.Nonnull; +import javax.annotation.Nullable; +import java.io.IOException; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +/** + * A {@link BaseTransformer} that processes input rows through a reusable scan query cursor pipeline. + * + * <p>The pipeline is built once at construction: a {@link SettableRowCursorFactory} is wrapped by the + * scan query's {@link SegmentMapFunction} (e.g., unnest, filter). For each input row, the row is set + * on the factory and the cursor is {@link Cursor#reset reset} — no per-row segment or cursor allocation. + * + * <p>When the scan query produces zero output rows (e.g., null/missing arrays, or filter rejection), + * the input row is dropped. This matches native Druid UNNEST / CROSS JOIN semantics where + * null or empty arrays produce zero rows. + * + * <p>This class is not thread-safe. Each reader thread should have its own instance. + */ +public class ScanTransformer implements BaseTransformer +{ + private final ScanQuery query; + // Field names that must never be promoted to a dimension, regardless of whether they show up in this + // scan query's result columns. See resolveDimensionColumns() for why this can't be derived locally. + private final Set<String> dimensionExclusions; + private final SettableRowCursorFactory baseCursorFactory; + private final CursorHolder cursorHolder; + private Cursor cursor; + + ScanTransformer(final ScanQuery scanQuery, final Set<String> dimensionExclusions) + { + this.query = scanQuery.withOverriddenContext( + Map.of(QueryContexts.TIMEOUT_KEY, 0) + ); + this.dimensionExclusions = dimensionExclusions; + + final RowSignature broadSignature = RowSignature.builder() + .add(ColumnHolder.TIME_COLUMN_NAME, ColumnType.LONG) + .build(); + + final CursorBuildSpec cursorBuildSpec = CursorBuildSpec.builder() + .setInterval(query.getSingleInterval()) + .setFilter(Filters.toFilter(query.getFilter())) + .setVirtualColumns(query.getVirtualColumns()) + .build(); + + this.baseCursorFactory = new SettableRowCursorFactory(broadSignature); + final SegmentMapFunction segmentMapFunction = query.getDataSource().createSegmentMapFunction(query); + final Segment mappedSegment = segmentMapFunction.apply(Optional.of(new CursorFactorySegment(baseCursorFactory))) + .orElseThrow(() -> new ISE("SegmentMapFunction returned empty")); + final CursorFactory mappedCursorFactory = mappedSegment.as(CursorFactory.class); + this.cursorHolder = mappedCursorFactory.makeCursorHolder(cursorBuildSpec); + } + + @Override + public boolean hasMultiRowTransform() + { + return true; + } + + @Override + @Nullable + public InputRow transform(@Nullable final InputRow row) + { + throw new UnsupportedOperationException( + "ScanTransformer does not support single-row transform; use transformToList()" + ); + } + + @Override + public List<InputRow> transformToList(@Nullable final InputRow row) + { + if (row == null) { + return List.of(); + } + + return process(row); + } + + @Override + @Nullable + public InputRowListPlusRawValues transform(@Nullable final InputRowListPlusRawValues row) + { + if (row == null || row.getInputRows() == null) { + return row; + } + + final List<InputRow> inputRows = row.getInputRows(); + final List<Map<String, Object>> inputRawValues = row.getRawValuesList(); + final List<InputRow> outputRows = new ArrayList<>(); + final List<Map<String, Object>> outputRawValues = inputRawValues == null ? null : new ArrayList<>(); + + for (int i = 0; i < inputRows.size(); i++) { + final List<InputRow> expandedRows = transformToList(inputRows.get(i)); + outputRows.addAll(expandedRows); + if (outputRawValues != null) { + for (int j = 0; j < expandedRows.size(); j++) { + outputRawValues.add(inputRawValues.get(i)); + } + } + } + + return InputRowListPlusRawValues.ofList(outputRawValues, outputRows, row.getParseException()); + } + + @Override + public void close() throws IOException + { + cursorHolder.close(); + } + + private List<InputRow> process(final InputRow inputRow) + { + baseCursorFactory.set(inputRow); + + if (cursor == null) { + cursor = cursorHolder.asCursor(); + } else { + cursor.reset(); + } + + if (cursor == null || cursor.isDone()) { + return List.of(); + } + + final Set<String> nonDimensionEventFields = resolveNonDimensionEventFields(inputRow); + final List<String> columns = resolveColumnsForRow(inputRow, nonDimensionEventFields); + final List<String> dimensionColumns = resolveDimensionColumns(inputRow, columns, nonDimensionEventFields); + final ColumnSelectorFactory selectorFactory = cursor.getColumnSelectorFactory(); + + // Selectors are lazy views over the cursor's current position — create them once per column + // here, then re-read via getObject() as the cursor advances, rather than reallocating a selector + // for every (output-row x column) pair. + final ColumnValueSelector<?>[] selectors = new ColumnValueSelector<?>[columns.size()]; + for (int i = 0; i < columns.size(); i++) { + selectors[i] = selectorFactory.makeColumnValueSelector(columns.get(i)); + } + + // The query re-executes fresh for each input row (cursor reset above), so this row's own unnested + // expansion is the query's entire result set for this execution — offset/limit bound that set, + // the same way they'd bound any other scan query's result set. This is necessarily per input row + // rather than global across the ingestion job: there is no single ordered stream spanning rows + // (let alone across the parallel/rolling readers of a real ingestion job) for them to paginate. + final long offset = query.getScanRowsOffset(); + final long limit = query.getScanRowsLimit(); + long skipped = 0; + + final List<InputRow> result = new ArrayList<>(); + while (!cursor.isDone() && result.size() < limit) { Review Comment: ## CodeQL / Comparison of narrow type with wide type in loop condition Comparison between [expression](1) of type int and [expression](2) of wider type long. [Show more details](https://github.com/apache/druid/security/code-scanning/11998) ########## processing/src/test/java/org/apache/druid/segment/transform/ScanTransformTest.java: ########## @@ -0,0 +1,717 @@ +/* + * 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.segment.transform; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.base.Preconditions; +import org.apache.druid.data.input.InputRow; +import org.apache.druid.data.input.ListBasedInputRow; +import org.apache.druid.data.input.MapBasedInputRow; +import org.apache.druid.error.DruidExceptionMatcher; +import org.apache.druid.java.util.common.DateTimes; +import org.apache.druid.math.expr.ExprMacroTable; +import org.apache.druid.query.Druids; +import org.apache.druid.query.TableDataSource; +import org.apache.druid.query.UnnestDataSource; +import org.apache.druid.query.filter.SelectorDimFilter; +import org.apache.druid.query.scan.ScanQuery; +import org.apache.druid.segment.TestHelper; +import org.apache.druid.segment.column.ColumnHolder; +import org.apache.druid.segment.column.ColumnType; +import org.apache.druid.segment.column.RowSignature; +import org.apache.druid.segment.virtual.ExpressionVirtualColumn; +import org.apache.druid.testing.InitializedNullHandlingTest; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +public class ScanTransformTest extends InitializedNullHandlingTest +{ + private static final long TIMESTAMP = DateTimes.of("2024-01-01").getMillis(); + + private static InputRow makeRow(Object... kvPairs) + { + Preconditions.checkArgument(kvPairs.length % 2 == 0, "kvPairs must have even length"); + final LinkedHashMap<String, Object> event = new LinkedHashMap<>(); + final List<String> dimensions = new ArrayList<>(); + for (int i = 0; i < kvPairs.length; i += 2) { + final String key = (String) kvPairs[i]; + event.put(key, kvPairs[i + 1]); + if (!ColumnHolder.TIME_COLUMN_NAME.equals(key)) { + dimensions.add(key); + } + } + return new MapBasedInputRow(TIMESTAMP, dimensions, event); + } + + private static ScanQuery makeUnnestQuery(String inputColumn, String outputName) + { + return makeUnnestQuery(inputColumn, outputName, ColumnType.STRING, null); + } + + private static ScanQuery makeUnnestQuery( + String inputColumn, + String outputName, + ColumnType outputType, + SelectorDimFilter unnestFilter + ) + { + return Druids.newScanQueryBuilder() + .dataSource(UnnestDataSource.create( + new TableDataSource("__input__"), + new ExpressionVirtualColumn(outputName, "\"" + inputColumn + "\"", outputType, ExprMacroTable.nil()), + unnestFilter + )) + .eternityInterval() + .columns((List<String>) null) + .resultFormat(ScanQuery.ResultFormat.RESULT_FORMAT_LIST) + .build(); + } + + private static BaseTransformer makeTransformer(ScanQuery query) + { + return new ScanTransformSpec(query).toTransformer(); + } + + @Test + public void testBasicUnnest() + { + final BaseTransformer transformer = makeTransformer(makeUnnestQuery("tags", "tag")); + final InputRow input = makeRow("user", "alice", "tags", List.of("a", "b", "c")); + + final List<InputRow> result = transformer.transformToList(input); + Assertions.assertEquals(3, result.size()); + + Assertions.assertEquals("a", result.get(0).getRaw("tag")); + Assertions.assertEquals("alice", result.get(0).getRaw("user")); + Assertions.assertEquals(TIMESTAMP, result.get(0).getTimestampFromEpoch()); + + Assertions.assertEquals("b", result.get(1).getRaw("tag")); + Assertions.assertEquals("c", result.get(2).getRaw("tag")); + } + + @Test + public void testUnnestEmptyArray() + { + final BaseTransformer transformer = makeTransformer(makeUnnestQuery("tags", "tag")); + final InputRow input = makeRow("user", "alice", "tags", List.of()); + + final List<InputRow> result = transformer.transformToList(input); + // Empty array produces 0 rows, matching native CROSS JOIN UNNEST semantics + Assertions.assertEquals(0, result.size()); + } + + @Test + public void testUnnestMissingColumn() + { + final BaseTransformer transformer = makeTransformer(makeUnnestQuery("services", "svc")); + final InputRow input = makeRow("user", "alice", "host", "web-01"); + + final List<InputRow> result = transformer.transformToList(input); + // Missing column produces 0 rows, matching native CROSS JOIN UNNEST semantics + Assertions.assertEquals(0, result.size()); + } + + @Test + public void testUnnestSingleElement() + { + final BaseTransformer transformer = makeTransformer(makeUnnestQuery("tags", "tag")); + final InputRow input = makeRow("user", "alice", "tags", List.of("only")); + + final List<InputRow> result = transformer.transformToList(input); + Assertions.assertEquals(1, result.size()); + Assertions.assertEquals("only", result.get(0).getRaw("tag")); + } + + @Test + public void testUnnestScalarValue() + { + final BaseTransformer transformer = makeTransformer(makeUnnestQuery("tags", "tag")); + final InputRow input = makeRow("user", "alice", "tags", "scalar"); + + final List<InputRow> result = transformer.transformToList(input); + Assertions.assertEquals(1, result.size()); + Assertions.assertEquals("scalar", result.get(0).getRaw("tag")); + } + + @Test + public void testUnnestArrayOfJsonObjects() + { + final BaseTransformer transformer = makeTransformer( + makeUnnestQuery("items", "item", ColumnType.NESTED_DATA, null) + ); + final InputRow input = makeRow("user", "alice", "items", List.of( + Map.of("product", "shirt", "price", 25), + Map.of("product", "pants", "price", 40), + Map.of("product", "hat", "price", 15) + )); + + final List<InputRow> result = transformer.transformToList(input); + Assertions.assertEquals(3, result.size()); + + final Object item0 = result.get(0).getRaw("item"); + Assertions.assertNotNull(item0); + Assertions.assertTrue(item0 instanceof Map, "Expected a Map, got " + item0.getClass()); + Assertions.assertEquals("shirt", ((Map<?, ?>) item0).get("product")); + + final Object item2 = result.get(2).getRaw("item"); + Assertions.assertTrue(item2 instanceof Map); + Assertions.assertEquals("hat", ((Map<?, ?>) item2).get("product")); + } + + @Test + public void testUnnestNestedArrays() + { + final BaseTransformer transformer = makeTransformer( + makeUnnestQuery("data", "element", ColumnType.NESTED_DATA, null) + ); + final InputRow input = makeRow( + "user", "alice", + "data", List.of(List.of(1, 2), List.of(3)) + ); + + final List<InputRow> result = transformer.transformToList(input); + Assertions.assertEquals(2, result.size()); + + final Object elem0 = result.get(0).getRaw("element"); + Assertions.assertNotNull(elem0); + Assertions.assertArrayEquals(new Object[]{1L, 2L}, (Object[]) elem0); + + final Object elem1 = result.get(1).getRaw("element"); + Assertions.assertNotNull(elem1); + Assertions.assertArrayEquals(new Object[]{3L}, (Object[]) elem1); + + Assertions.assertEquals("alice", result.get(0).getRaw("user")); + Assertions.assertEquals("alice", result.get(1).getRaw("user")); + } + + @Test + public void testTimestampPreservation() + { + final BaseTransformer transformer = makeTransformer(makeUnnestQuery("tags", "tag")); + final InputRow input = makeRow("tags", List.of("a", "b")); + + final List<InputRow> result = transformer.transformToList(input); + for (final InputRow row : result) { + Assertions.assertEquals(TIMESTAMP, row.getTimestampFromEpoch()); + } + } + + @Test + public void testWithUnnestFilter() + { + final BaseTransformer transformer = makeTransformer( + makeUnnestQuery("tags", "tag", ColumnType.STRING, new SelectorDimFilter("tag", "b", null)) + ); + final InputRow input = makeRow("user", "alice", "tags", List.of("a", "b", "c")); + + final List<InputRow> result = transformer.transformToList(input); + Assertions.assertEquals(1, result.size()); + Assertions.assertEquals("b", result.get(0).getRaw("tag")); + } + + @Test + public void testScanQueryLimitCapsPerRowExpansion() + { + // The query re-executes fresh for each input row, so this row's own unnested elements are the + // query's entire result set for that execution — limit:1 caps it to the first unnested element. + final ScanQuery query = Druids.newScanQueryBuilder() + .dataSource(UnnestDataSource.create( + new TableDataSource("__input__"), + new ExpressionVirtualColumn("tag", "\"tags\"", ColumnType.STRING, ExprMacroTable.nil()), + null + )) + .eternityInterval() + .columns((List<String>) null) + .resultFormat(ScanQuery.ResultFormat.RESULT_FORMAT_LIST) + .limit(1) + .build(); + + final BaseTransformer transformer = new ScanTransformSpec(query).toTransformer(); + final InputRow input = makeRow("user", "alice", "tags", List.of("a", "b", "c")); + + final List<InputRow> result = transformer.transformToList(input); + Assertions.assertEquals(1, result.size()); + Assertions.assertEquals("a", result.get(0).getRaw("tag")); + } + + @Test + public void testScanQueryOffsetSkipsPerRowExpansion() + { + final ScanQuery query = Druids.newScanQueryBuilder() + .dataSource(UnnestDataSource.create( + new TableDataSource("__input__"), + new ExpressionVirtualColumn("tag", "\"tags\"", ColumnType.STRING, ExprMacroTable.nil()), + null + )) + .eternityInterval() + .columns((List<String>) null) + .resultFormat(ScanQuery.ResultFormat.RESULT_FORMAT_LIST) + .offset(1) + .limit(1) + .build(); + + final BaseTransformer transformer = new ScanTransformSpec(query).toTransformer(); + final InputRow input = makeRow("user", "alice", "tags", List.of("a", "b", "c")); + + final List<InputRow> result = transformer.transformToList(input); + Assertions.assertEquals(1, result.size()); + Assertions.assertEquals("b", result.get(0).getRaw("tag")); + } + + @Test + public void testScanQueryLimitAppliesIndependentlyPerInputRow() + { + // limit is bound to each execution of the query, i.e. each input row's own expansion — not a + // running total across the multiple input rows a reader processes over its lifetime. + final ScanQuery query = Druids.newScanQueryBuilder() + .dataSource(UnnestDataSource.create( + new TableDataSource("__input__"), + new ExpressionVirtualColumn("tag", "\"tags\"", ColumnType.STRING, ExprMacroTable.nil()), + null + )) + .eternityInterval() + .columns((List<String>) null) + .resultFormat(ScanQuery.ResultFormat.RESULT_FORMAT_LIST) + .limit(1) + .build(); + + final BaseTransformer transformer = new ScanTransformSpec(query).toTransformer(); + + final List<InputRow> firstResult = transformer.transformToList( + makeRow("user", "alice", "tags", List.of("a", "b")) + ); + final List<InputRow> secondResult = transformer.transformToList( + makeRow("user", "bob", "tags", List.of("x", "y")) + ); + + Assertions.assertEquals(1, firstResult.size()); + Assertions.assertEquals("a", firstResult.get(0).getRaw("tag")); + Assertions.assertEquals(1, secondResult.size()); + Assertions.assertEquals("x", secondResult.get(0).getRaw("tag")); + } + + @Test + public void testTransformerWithSingleScanTransform() + { + final ScanTransformSpec spec = new ScanTransformSpec( + makeUnnestQuery("tags", "tag") + ); + + final BaseTransformer transformer = spec.toTransformer(); + Assertions.assertTrue(transformer.hasMultiRowTransform()); + Assertions.assertTrue(transformer instanceof ScanTransformer); + + final InputRow input = makeRow("user", "alice", "tags", List.of("x", "y")); + final List<InputRow> result = transformer.transformToList(input); + + Assertions.assertEquals(2, result.size()); + Assertions.assertEquals("x", result.get(0).getRaw("tag")); + Assertions.assertEquals("y", result.get(1).getRaw("tag")); + } + + @Test + public void testNestedUnnestCrossJoin() + { + final BaseTransformer transformer = new ScanTransformSpec( + Druids.newScanQueryBuilder() + .dataSource(UnnestDataSource.create( + UnnestDataSource.create( + new TableDataSource("__input__"), + new ExpressionVirtualColumn("tag", "\"tags\"", ColumnType.STRING, ExprMacroTable.nil()), + null + ), + new ExpressionVirtualColumn("color", "\"colors\"", ColumnType.STRING, ExprMacroTable.nil()), + null + )) + .eternityInterval() + .columns((List<String>) null) + .resultFormat(ScanQuery.ResultFormat.RESULT_FORMAT_LIST) + .build() + ).toTransformer(); + Assertions.assertTrue(transformer.hasMultiRowTransform()); + + final InputRow input = makeRow( + "user", "alice", + "tags", List.of("a", "b"), + "colors", List.of("red", "blue", "green") + ); + final List<InputRow> result = transformer.transformToList(input); + + // 2 tags x 3 colors = 6 rows (cross join) + Assertions.assertEquals(6, result.size()); + } + + @Test + public void testNestedUnnestWithMissingOuterColumn() + { + final BaseTransformer transformer = new ScanTransformSpec( + Druids.newScanQueryBuilder() + .dataSource(UnnestDataSource.create( + UnnestDataSource.create( + new TableDataSource("__input__"), + new ExpressionVirtualColumn("tag", "\"tags\"", ColumnType.STRING, ExprMacroTable.nil()), + null + ), + new ExpressionVirtualColumn("svc", "\"services\"", ColumnType.NESTED_DATA, ExprMacroTable.nil()), + null + )) + .eternityInterval() + .columns((List<String>) null) + .resultFormat(ScanQuery.ResultFormat.RESULT_FORMAT_LIST) + .build() + ).toTransformer(); + + // tags present (2 elements), services missing + final InputRow input = makeRow("trace_id", "abc", "tags", List.of("music", "blll")); + final List<InputRow> result = transformer.transformToList(input); + + // Nested unnest is a cross join: tags x services. With services missing, the cross join + // produces 0 rows — matching native CROSS JOIN UNNEST semantics. + Assertions.assertEquals(0, result.size()); + } + + @Test + public void testNestedUnnestFlattensNestedArrays() + { + final BaseTransformer transformer = new ScanTransformSpec( + Druids.newScanQueryBuilder() + .dataSource(UnnestDataSource.create( + UnnestDataSource.create( + new TableDataSource("__input__"), + new ExpressionVirtualColumn("inner", "\"data\"", ColumnType.NESTED_DATA, ExprMacroTable.nil()), + null + ), + new ExpressionVirtualColumn("val", "\"inner\"", ColumnType.LONG, ExprMacroTable.nil()), + null + )) + .eternityInterval() + .columns((List<String>) null) + .resultFormat(ScanQuery.ResultFormat.RESULT_FORMAT_LIST) + .build() + ).toTransformer(); + + final InputRow input = makeRow( + "user", "alice", + "data", List.of(List.of(1, 2), List.of(3)) + ); + final List<InputRow> result = transformer.transformToList(input); + + // First unnest: [[1,2],[3]] -> [1,2], [3] (2 rows) + // Second unnest: [1,2] -> 1, 2 and [3] -> 3 (3 rows total) + Assertions.assertEquals(3, result.size()); + + final List<Object> values = new ArrayList<>(); + for (final InputRow row : result) { + values.add(row.getRaw("val")); + Assertions.assertEquals("alice", row.getRaw("user")); + } + Assertions.assertEquals(3, values.size()); + Assertions.assertEquals(1, ((Number) values.get(0)).intValue()); + Assertions.assertEquals(2, ((Number) values.get(1)).intValue()); + Assertions.assertEquals(3, ((Number) values.get(2)).intValue()); + } + + @Test + public void testScanTransformWithQueryFilter() + { + final BaseTransformer transformer = new ScanTransformSpec( + Druids.newScanQueryBuilder() + .dataSource(UnnestDataSource.create( + new TableDataSource("__input__"), + new ExpressionVirtualColumn("tag", "\"tags\"", ColumnType.STRING, ExprMacroTable.nil()), + null + )) + .eternityInterval() + .filters(new SelectorDimFilter("user", "not_alice", null)) + .columns((List<String>) null) + .resultFormat(ScanQuery.ResultFormat.RESULT_FORMAT_LIST) + .build() + ).toTransformer(); + final InputRow input = makeRow("user", "alice", "tags", List.of("a", "b")); + final List<InputRow> result = transformer.transformToList(input); + // Filter rejects the row (user != "not_alice"), so 0 rows — matching native scan query semantics + Assertions.assertEquals(0, result.size()); + } + + @Test + public void testTransformerWithoutScanTransform() + { + final TransformSpec spec = new TransformSpec(null, null); + final BaseTransformer transformer = spec.toTransformer(); + Assertions.assertFalse(transformer.hasMultiRowTransform()); + Assertions.assertTrue(transformer instanceof Transformer); + + final InputRow input = makeRow("user", "alice"); + final List<InputRow> result = transformer.transformToList(input); + Assertions.assertEquals(1, result.size()); + } + + @Test + public void testTransformerTransformToListWithNull() + { + final TransformSpec spec = new TransformSpec(null, null); + final BaseTransformer transformer = spec.toTransformer(); + Assertions.assertTrue(transformer.transformToList(null).isEmpty()); + } + + // --- Serde tests --- + + @Test + public void testScanTransformSpecSerde() throws Exception + { + final ScanTransformSpec spec = new ScanTransformSpec( + makeUnnestQuery("tags", "tag", ColumnType.STRING, new SelectorDimFilter("tag", "a", null)) + ); + + final ObjectMapper jsonMapper = TestHelper.makeJsonMapper(); + final String json = jsonMapper.writeValueAsString(spec); + final BaseTransformSpec deserialized = jsonMapper.readValue(json, BaseTransformSpec.class); + Assertions.assertTrue(deserialized instanceof ScanTransformSpec); + Assertions.assertEquals(spec, deserialized); + } + + @Test + public void testUnnestPreservesNonDimensionEventFields() + { + // Simulates a fixed-dimensions ingestion with a metric: `bytes_sent` is in the raw event map + // but excluded from getDimensions() because DataSchema added it to dimensionExclusions. + // The expanded rows must still carry `bytes_sent` so downstream aggregators can read it, but must + // NOT promote it to a dimension — IncrementalIndex treats getDimensions() as authoritative and + // would otherwise auto-discover/store the metric source field as a dimension. + final LinkedHashMap<String, Object> event = new LinkedHashMap<>(); + event.put("user", "alice"); + event.put("tags", List.of("a", "b")); + event.put("bytes_sent", 1024L); + final InputRow input = new MapBasedInputRow(TIMESTAMP, List.of("user", "tags"), event); + + final BaseTransformer transformer = makeTransformer(makeUnnestQuery("tags", "tag")); + final List<InputRow> result = transformer.transformToList(input); + + Assertions.assertEquals(2, result.size()); + for (final InputRow row : result) { + Assertions.assertEquals(1024L, row.getRaw("bytes_sent")); + Assertions.assertEquals("alice", row.getRaw("user")); + Assertions.assertFalse( + row.getDimensions().contains("bytes_sent"), + "metric source field must not be promoted to a dimension" + ); + Assertions.assertTrue(row.getDimensions().contains("user")); + Assertions.assertTrue(row.getDimensions().contains("tag")); + } + Assertions.assertEquals("a", result.get(0).getRaw("tag")); + Assertions.assertEquals("b", result.get(1).getRaw("tag")); + } + + @Test + public void testUnnestDoesNotDiscoverMetricSourceFieldAsDimension() + { + // Same scenario as above, but with multiple metric source fields (bytes_sent, latency_ms) that + // are never referenced by dimensions, virtual columns, or unnest output. None of them should + // appear in the expanded rows' dimensions. + final LinkedHashMap<String, Object> event = new LinkedHashMap<>(); + event.put("host", "web-01"); + event.put("tags", List.of("x", "y", "z")); + event.put("bytes_sent", 2048L); + event.put("latency_ms", 42L); + final InputRow input = new MapBasedInputRow(TIMESTAMP, List.of("host", "tags"), event); + + final BaseTransformer transformer = makeTransformer(makeUnnestQuery("tags", "tag")); + final List<InputRow> result = transformer.transformToList(input); + + Assertions.assertEquals(3, result.size()); + for (final InputRow row : result) { + Assertions.assertEquals(2048L, row.getRaw("bytes_sent")); + Assertions.assertEquals(42L, row.getRaw("latency_ms")); + Assertions.assertEquals(List.of("host", "tags", "tag"), row.getDimensions()); + Assertions.assertFalse(row.getDimensions().contains("bytes_sent")); + Assertions.assertFalse(row.getDimensions().contains("latency_ms")); + } + } + + @Test + public void testGeneratedVirtualColumnOutputExcludedFromDimensions() + { + // "rate" is a virtual column consumed only by an aggregator, so DataSchema would put it in + // dimensionExclusions. Since it's generated by the scan query (not a raw event field), it can only + // be kept out of the dimension list if ScanTransformer is told about the exclusion explicitly. + final ScanQuery query = Druids.newScanQueryBuilder() + .dataSource(new TableDataSource("__input__")) + .virtualColumns(new ExpressionVirtualColumn( + "rate", + "\"total\" / 2", + ColumnType.LONG, + ExprMacroTable.nil() + )) + .eternityInterval() + .columns((List<String>) null) + .resultFormat(ScanQuery.ResultFormat.RESULT_FORMAT_LIST) + .build(); + + final BaseTransformer transformer = new ScanTransformSpec(query).toTransformer(Set.of("rate")); + final InputRow input = makeRow("user", "alice", "total", 10L); + + final List<InputRow> result = transformer.transformToList(input); + Assertions.assertEquals(1, result.size()); + Assertions.assertEquals(5L, result.get(0).getRaw("rate")); + Assertions.assertFalse( + result.get(0).getDimensions().contains("rate"), + "metric-only virtual column output must not be promoted to a dimension" + ); + Assertions.assertTrue(result.get(0).getDimensions().contains("user")); + Assertions.assertTrue(result.get(0).getDimensions().contains("total")); + } + + @Test + public void testGeneratedVirtualColumnOutputPromotedToDimensionWithoutExclusions() + { + // Same query as above, but via the no-arg toTransformer() (no dimensionExclusions supplied). + // Without exclusion info, "rate" is promoted to a dimension — this documents the pre-existing behavior. + final ScanQuery query = Druids.newScanQueryBuilder() + .dataSource(new TableDataSource("__input__")) + .virtualColumns(new ExpressionVirtualColumn( + "rate", + "\"total\" / 2", + ColumnType.LONG, + ExprMacroTable.nil() + )) + .eternityInterval() + .columns((List<String>) null) + .resultFormat(ScanQuery.ResultFormat.RESULT_FORMAT_LIST) + .build(); + + final BaseTransformer transformer = new ScanTransformSpec(query).toTransformer(); + final InputRow input = makeRow("user", "alice", "total", 10L); + + final List<InputRow> result = transformer.transformToList(input); + Assertions.assertEquals(1, result.size()); + Assertions.assertTrue(result.get(0).getDimensions().contains("rate")); + } + + @Test + public void testListBasedInputRowPreservesNonDimensionFieldsThroughUnnest() + { + // CSV/TSV/delimited ingestion produces ListBasedInputRow, not MapBasedInputRow. A metric-only field + // (bytes_sent) must still be carried into the expanded rows' event maps, same as for MapBasedInputRow. + final RowSignature signature = RowSignature.builder() + .add("user", ColumnType.STRING) + .add("tags", ColumnType.STRING_ARRAY) + .add("bytes_sent", ColumnType.LONG) + .build(); + final InputRow input = new ListBasedInputRow( + signature, + DateTimes.utc(TIMESTAMP), + List.of("user", "tags"), + List.of("alice", List.of("a", "b"), 1024L) + ); + + final BaseTransformer transformer = makeTransformer(makeUnnestQuery("tags", "tag")); + final List<InputRow> result = transformer.transformToList(input); + + Assertions.assertEquals(2, result.size()); + for (final InputRow row : result) { + Assertions.assertEquals(1024L, row.getRaw("bytes_sent")); + Assertions.assertEquals("alice", row.getRaw("user")); + Assertions.assertFalse( + row.getDimensions().contains("bytes_sent"), + "metric source field must not be promoted to a dimension for list-backed rows either" + ); + } + Assertions.assertEquals("a", result.get(0).getRaw("tag")); + Assertions.assertEquals("b", result.get(1).getRaw("tag")); + } + + @Test + public void testUnsupportedInputRowTypeThrowsDefensiveException() + { + // Extension InputRow implementations (e.g. DeltaInputRow from druid-deltalake-extensions) implement + // InputRow directly rather than extending MapBasedInputRow/ListBasedInputRow, and have no generic + // Map-shaped view of their raw fields. ScanTransformer has no way to preserve non-dimension event + // fields (e.g. metrics) for such rows, so it must fail loudly instead of silently dropping them. + final InputRow input = new OpaqueInputRow(makeRow("user", "alice", "tags", List.of("a", "b"))); + final BaseTransformer transformer = makeTransformer(makeUnnestQuery("tags", "tag")); + + DruidExceptionMatcher.defensive().expectMessageContains( + "ScanTransformer does not support input rows of type" + ).assertThrowsAndMatches(() -> transformer.transformToList(input)); + } + + /** + * A minimal {@link InputRow} that delegates to another row without extending + * {@link MapBasedInputRow} or {@link ListBasedInputRow} — simulates an extension-provided row type + * such as {@code DeltaInputRow}, which has no generic Map-shaped view of its raw fields. + */ + private static class OpaqueInputRow implements InputRow Review Comment: ## CodeQL / Inconsistent compareTo This class declares [compareTo](1) but inherits equals; the two could be inconsistent. [Show more details](https://github.com/apache/druid/security/code-scanning/11999) -- 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]
