You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "clintropolis (via GitHub)" <gi...@apache.org> on 2026/06/13 18:50:23 UTC
[PR] feat: clustered segments pt.2 (write side support) (druid)
clintropolis opened a new pull request, #19579:
URL: https://github.com/apache/druid/pull/19579
### Description
Follow-up to #19460, this PR introduces the writer side stuff so that the segments can actually be created.
changes:
* adds `BaseTableProjectionSpec` interface to capture the operator facing shape of V10 base table schemas
* adds `ClusteredValueGroupsBaseTableSchema` implementation for ingesting clustered segments
* adds `DataSchema.baseTable`, a `BaseTableProjectionSpec` which when set puts the DataSchema into a new mode where the majority of the schema is defined via the baseTable, rejecting other top level fields
* adds `DataSchema.segmentGranularity` to use when `baseTable` is set, which captures the segment granularity and intervals (query granularity is defined by the baseTable)
* adds `AdaptedBaseTableProjectionSpec` implementation for converting classic `DataSchema` fields to a `BaseTableProjectionSpec`
* adds `OnHeapClusteredBaseTable`, `OnHeapClusterGroup` used by `OnHeapIncrementalIndex` to build clustered segments
* adds `IndexMergerV10.makeClusteredIndexFiles` which merges and builds clustered v10 segments
* `Sink`/`BatchAppenderator`/`StreamAppenderator` wiring for clustered segments so that the cluster group tuples appear on the `DataSegment`
* known issues: unbounded, no aggregate projections, no compaction support, no time ordered cursor support
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: clustered segments pt.2 (write side support) (druid)
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis commented on code in PR #19579:
URL: https://github.com/apache/druid/pull/19579#discussion_r3415364782
##########
processing/src/main/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpec.java:
##########
@@ -0,0 +1,298 @@
+/*
+ * 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.impl;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import com.google.common.collect.Sets;
+import org.apache.druid.error.InvalidInput;
+import org.apache.druid.query.OrderBy;
+import org.apache.druid.query.aggregation.AggregatorFactory;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.column.ColumnHolder;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * {@link BaseTableProjectionSpec} for the clustered-value-groups base table mode: rows are partitioned into per-tuple
+ * "cluster groups" keyed by one or more typed clustering columns, optionally derived from {@link #virtualColumns}.
+ * Essentially, each group is stored as its own internal sub-segment..
+ * <p>
+ * The operator declares (1) the typed clustering columns (as ordinary {@link DimensionSchema}s, they're real
+ * dimensions, just stored differently), (2) optional virtual columns, and (3) the non-clustering
+ * dimensions and metrics. {@link #getDimensionsSpec()} returns the unified spec in canonical order
+ * (clustering dims first, then non-clustering); {@link #getOrdering()} is either the declared ordering or, by
+ * default, all clustering dims ascending followed by {@code __time} ascending.
+ * <p>
+ * A clustered base table is never rollup. Query granularity, when wanted, is just another entry in
+ * {@link #getVirtualColumns()} named
+ * {@link org.apache.druid.java.util.common.granularity.Granularities#GRANULARITY_VIRTUAL_COLUMN_NAME}; absent that
+ * virtual column the query granularity is {@code NONE}. Segment granularity and intervals live on the top-level
+ * {@link org.apache.druid.indexer.granularity.SegmentGranularitySpec}, not here.
+ */
+@JsonTypeName(ClusteredValueGroupsBaseTableProjectionSpec.TYPE_NAME)
+public final class ClusteredValueGroupsBaseTableProjectionSpec implements BaseTableProjectionSpec
+{
+ public static final String TYPE_NAME = "clusteredValueGroups";
+
+ private final VirtualColumns virtualColumns;
+ private final List<DimensionSchema> clusteringColumns;
+ private final List<DimensionSchema> nonClusteringDimensions;
+ private final AggregatorFactory[] metrics;
+ private final DimensionsSpec dimensionsSpec;
+ private final List<OrderBy> ordering;
+
+ public static Builder builder()
+ {
+ return new Builder();
+ }
+
+ @JsonCreator
+ public ClusteredValueGroupsBaseTableProjectionSpec(
+ @JsonProperty("virtualColumns") @Nullable VirtualColumns virtualColumns,
+ @JsonProperty("clusteringColumns") List<DimensionSchema> clusteringColumns,
+ @JsonProperty("dimensions") @Nullable List<DimensionSchema> nonClusteringDimensions,
+ @JsonProperty("metrics") @Nullable AggregatorFactory[] metrics,
+ @JsonProperty("ordering") @Nullable List<OrderBy> declaredOrdering
+ )
+ {
+ if (clusteringColumns == null || clusteringColumns.isEmpty()) {
+ throw InvalidInput.exception("clusteringColumns must be non-empty for clusteredValueGroups base table");
+ }
+ this.clusteringColumns = Collections.unmodifiableList(new ArrayList<>(clusteringColumns));
+ this.virtualColumns = virtualColumns == null ? VirtualColumns.EMPTY : virtualColumns;
+ this.nonClusteringDimensions = nonClusteringDimensions == null
+ ? Collections.emptyList()
+ : Collections.unmodifiableList(new ArrayList<>(nonClusteringDimensions));
+ this.metrics = metrics == null ? new AggregatorFactory[0] : metrics;
+ validateNoOverlap(this.clusteringColumns, this.nonClusteringDimensions);
+ this.dimensionsSpec = computeDimensionsSpec(this.clusteringColumns, this.nonClusteringDimensions);
+ this.ordering = declaredOrdering != null
Review Comment:
ah this is fair, i should probably make it purely computed for now like `AggregateProjectionSpec` since its a bit of work to actually wire this up to honor it
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: clustered segments pt.2 (write side support) (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #19579:
URL: https://github.com/apache/druid/pull/19579#discussion_r3413389507
##########
processing/src/main/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpec.java:
##########
@@ -0,0 +1,298 @@
+/*
+ * 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.impl;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import com.google.common.collect.Sets;
+import org.apache.druid.error.InvalidInput;
+import org.apache.druid.query.OrderBy;
+import org.apache.druid.query.aggregation.AggregatorFactory;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.column.ColumnHolder;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * {@link BaseTableProjectionSpec} for the clustered-value-groups base table mode: rows are partitioned into per-tuple
+ * "cluster groups" keyed by one or more typed clustering columns, optionally derived from {@link #virtualColumns}.
+ * Essentially, each group is stored as its own internal sub-segment..
+ * <p>
+ * The operator declares (1) the typed clustering columns (as ordinary {@link DimensionSchema}s, they're real
+ * dimensions, just stored differently), (2) optional virtual columns, and (3) the non-clustering
+ * dimensions and metrics. {@link #getDimensionsSpec()} returns the unified spec in canonical order
+ * (clustering dims first, then non-clustering); {@link #getOrdering()} is either the declared ordering or, by
+ * default, all clustering dims ascending followed by {@code __time} ascending.
+ * <p>
+ * A clustered base table is never rollup. Query granularity, when wanted, is just another entry in
+ * {@link #getVirtualColumns()} named
+ * {@link org.apache.druid.java.util.common.granularity.Granularities#GRANULARITY_VIRTUAL_COLUMN_NAME}; absent that
+ * virtual column the query granularity is {@code NONE}. Segment granularity and intervals live on the top-level
+ * {@link org.apache.druid.indexer.granularity.SegmentGranularitySpec}, not here.
+ */
+@JsonTypeName(ClusteredValueGroupsBaseTableProjectionSpec.TYPE_NAME)
+public final class ClusteredValueGroupsBaseTableProjectionSpec implements BaseTableProjectionSpec
+{
+ public static final String TYPE_NAME = "clusteredValueGroups";
+
+ private final VirtualColumns virtualColumns;
+ private final List<DimensionSchema> clusteringColumns;
+ private final List<DimensionSchema> nonClusteringDimensions;
+ private final AggregatorFactory[] metrics;
+ private final DimensionsSpec dimensionsSpec;
+ private final List<OrderBy> ordering;
+
+ public static Builder builder()
+ {
+ return new Builder();
+ }
+
+ @JsonCreator
+ public ClusteredValueGroupsBaseTableProjectionSpec(
+ @JsonProperty("virtualColumns") @Nullable VirtualColumns virtualColumns,
+ @JsonProperty("clusteringColumns") List<DimensionSchema> clusteringColumns,
+ @JsonProperty("dimensions") @Nullable List<DimensionSchema> nonClusteringDimensions,
+ @JsonProperty("metrics") @Nullable AggregatorFactory[] metrics,
+ @JsonProperty("ordering") @Nullable List<OrderBy> declaredOrdering
+ )
+ {
+ if (clusteringColumns == null || clusteringColumns.isEmpty()) {
+ throw InvalidInput.exception("clusteringColumns must be non-empty for clusteredValueGroups base table");
+ }
+ this.clusteringColumns = Collections.unmodifiableList(new ArrayList<>(clusteringColumns));
+ this.virtualColumns = virtualColumns == null ? VirtualColumns.EMPTY : virtualColumns;
+ this.nonClusteringDimensions = nonClusteringDimensions == null
+ ? Collections.emptyList()
+ : Collections.unmodifiableList(new ArrayList<>(nonClusteringDimensions));
+ this.metrics = metrics == null ? new AggregatorFactory[0] : metrics;
+ validateNoOverlap(this.clusteringColumns, this.nonClusteringDimensions);
+ this.dimensionsSpec = computeDimensionsSpec(this.clusteringColumns, this.nonClusteringDimensions);
+ this.ordering = declaredOrdering != null
Review Comment:
[P2] Declared ordering can be advertised without being honored
The spec accepts any declaredOrdering here, but the clustered write path does not use it to build the actual per-group row comparator: OnHeapClusterGroup derives ordering from timePosition/DimensionsSpec, while IndexMergerV10 emits groups in ascending tuple order and then stores firstSchema.getOrdering() in metadata. A spec such as [tenant ASC, page ASC, __time ASC], or any DESC order, can therefore be persisted with advertised ordering that the data does not actually satisfy. Query engines may trust CursorHolder/DataSegment ordering and skip sorting, producing misordered results. Please reject unsupported ordering or wire the declared order through ingestion and merge.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: clustered segments pt.2 (write side support) (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #19579:
URL: https://github.com/apache/druid/pull/19579#discussion_r3420712714
##########
processing/src/main/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpec.java:
##########
@@ -0,0 +1,345 @@
+/*
+ * 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.impl;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import com.google.common.collect.Sets;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.error.InvalidInput;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.query.OrderBy;
+import org.apache.druid.query.aggregation.AggregatorFactory;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.column.ColumnHolder;
+import org.apache.druid.utils.CollectionUtils;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * {@link BaseTableProjectionSpec} for the clustered-value-groups base table mode: rows are partitioned into per-tuple
+ * "cluster groups" keyed by one or more typed clustering columns, optionally derived from {@link #virtualColumns}.
+ * Essentially, each group is stored as its own internal sub-segment..
+ * <p>
+ * The operator declares a single ordered {@link #columns} list — the full set of columns in segment order, plus a
+ * {@link #clusteringColumns} list of NAMES designating the leading prefix of {@link #columns} that rows are
+ * clustered by. The time position is an explicit positional entry in {@link #columns} (named {@code __time}, or the
+ * query-granularity column {@link Granularities#GRANULARITY_VIRTUAL_COLUMN_NAME}); clustering by the time column is not
+ * yet supported, so the time marker must be a non-clustering column. A clustered base table is never rollup and has
+ * no metric columns.
+ * <p>
+ * {@link #getDimensionsSpec()} returns the unified spec built from {@link #columns} in declared order with
+ * {@code forceSegmentSortByTime=false}; {@link #getOrdering()} is computed as every column of {@link #columns}
+ * ascending, in list order.
+ * <p>
+ * Query granularity, when wanted, is just another entry in {@link #getVirtualColumns()} named
+ * {@link Granularities#GRANULARITY_VIRTUAL_COLUMN_NAME}; absent that virtual column the query granularity is
+ * {@code NONE}. Segment granularity and intervals live on the top-level
+ * {@link org.apache.druid.indexer.granularity.SegmentGranularitySpec}, not here.
+ */
+@JsonTypeName(ClusteredValueGroupsBaseTableProjectionSpec.TYPE_NAME)
+public final class ClusteredValueGroupsBaseTableProjectionSpec implements BaseTableProjectionSpec
+{
+ public static final String TYPE_NAME = "clusteredValueGroups";
+
+ private final VirtualColumns virtualColumns;
+ private final List<DimensionSchema> columns;
+ private final List<String> clusteringColumns;
+ private final List<DimensionSchema> clusteringColumnSchemas;
+ private final List<DimensionSchema> nonClusteringColumnSchemas;
+ private final DimensionsSpec dimensionsSpec;
+ private final List<OrderBy> ordering;
+
+ public static Builder builder()
+ {
+ return new Builder();
+ }
+
+ @JsonCreator
+ public ClusteredValueGroupsBaseTableProjectionSpec(
+ @JsonProperty("virtualColumns") @Nullable VirtualColumns virtualColumns,
+ @JsonProperty("columns") List<DimensionSchema> columns,
+ @JsonProperty("clusteringColumns") List<String> clusteringColumns
+ )
+ {
+ validate(columns, clusteringColumns);
+ this.virtualColumns = virtualColumns == null ? VirtualColumns.EMPTY : virtualColumns;
+ this.columns = Collections.unmodifiableList(new ArrayList<>(columns));
+ this.clusteringColumns = Collections.unmodifiableList(new ArrayList<>(clusteringColumns));
+ this.clusteringColumnSchemas = this.columns.subList(0, this.clusteringColumns.size());
+ this.nonClusteringColumnSchemas = this.columns.subList(this.clusteringColumns.size(), this.columns.size());
+ this.dimensionsSpec = DimensionsSpec.builder()
+ .setDimensions(this.columns)
+ .setForceSegmentSortByTime(false)
+ .build();
+ this.ordering = computeOrdering(this.columns);
+ }
+
+ @Override
+ @JsonProperty
+ @JsonInclude(JsonInclude.Include.NON_DEFAULT)
+ public VirtualColumns getVirtualColumns()
+ {
+ return virtualColumns;
+ }
+
+ /**
+ * The full, ordered list of columns in segment order; the clustering prefix followed by the remaining
+ * (non-clustering) columns, with the explicit {@code __time} (or query-granularity) marker at its declared
+ * position.
+ */
+ @JsonProperty("columns")
+ public List<DimensionSchema> getColumns()
+ {
+ return columns;
+ }
+
+ /**
+ * The names of the leading prefix of {@link #getColumns()} that rows are clustered by.
+ */
+ @JsonProperty("clusteringColumns")
+ public List<String> getClusteringColumnNames()
+ {
+ return clusteringColumns;
+ }
+
+ /**
+ * The clustering columns as {@link DimensionSchema}s; the leading prefix of {@link #getColumns()}.
+ */
+ @JsonIgnore
+ public List<DimensionSchema> getClusteringColumns()
+ {
+ return clusteringColumnSchemas;
+ }
+
+ /**
+ * The non-clustering columns as {@link DimensionSchema}s; the suffix of {@link #getColumns()} after the
+ * clustering prefix. Includes the explicit time marker.
+ */
+ @JsonIgnore
+ public List<DimensionSchema> getNonClusteringColumns()
+ {
+ return nonClusteringColumnSchemas;
+ }
+
+ @Override
+ public AggregatorFactory[] getMetrics()
+ {
+ return new AggregatorFactory[0];
+ }
+
+ @Override
+ @JsonIgnore
+ public List<OrderBy> getOrdering()
+ {
+ return ordering;
+ }
+
+ @Override
+ @JsonIgnore
+ public DimensionsSpec getDimensionsSpec()
+ {
+ return dimensionsSpec;
+ }
+
+ private static void validate(List<DimensionSchema> columns, List<String> clusteringColumns)
+ {
+ if (CollectionUtils.isNullOrEmpty(clusteringColumns)) {
+ throw InvalidInput.exception("clusteringColumns must be non-empty for clusteredValueGroups base table");
+ }
+ if (CollectionUtils.isNullOrEmpty(columns)) {
+ throw InvalidInput.exception("columns must be non-empty for clusteredValueGroups base table");
+ }
+ if (clusteringColumns.size() > columns.size()) {
+ throw clusteringPrefixException(columns, clusteringColumns);
+ }
+ for (int i = 0; i < clusteringColumns.size(); i++) {
+ if (!columns.get(i).getName().equals(clusteringColumns.get(i))) {
+ throw clusteringPrefixException(columns, clusteringColumns);
+ }
+ }
+
+ final Set<String> seen = Sets.newHashSetWithExpectedSize(columns.size());
+ for (DimensionSchema d : columns) {
+ if (!seen.add(d.getName())) {
+ throw InvalidInput.exception("columns contains duplicate name [%s]", d.getName());
+ }
+ }
+
+ int timeIndex = -1;
+ boolean bothPresent = false;
+ for (int i = 0; i < columns.size(); i++) {
+ final String name = columns.get(i).getName();
+ if (ColumnHolder.TIME_COLUMN_NAME.equals(name) || Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME.equals(name)) {
Review Comment:
[P2] Reject the virtual granularity time marker until it is wired through
This validation accepts a spec whose only time marker in `columns` is `Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME`, but the clustered write/read metadata path still assumes `ColumnHolder.TIME_COLUMN_NAME`: `IncrementalIndex` discovers `timePosition` only from `__time`, `OnHeapClusterGroup` skips only `__time` when building per-group dimensions, `BaseTableProjectionSchema#getTimeColumnName` defaults to `__time`, and `ClusteredValueGroupsBaseTableSchema` throws unless ordering contains `__time`. A clustered base-table spec with `__virtualGranularity` in columns and no `__time` therefore passes this validation, then either treats the marker as an ordinary dimension or fails during metadata schema construction. Please require `__time` for now, or wire the virtual-granularity marker through the clustered schema/group writer consistently.
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: clustered segments pt.2 (write side support) (druid)
Posted by "capistrant (via GitHub)" <gi...@apache.org>.
capistrant commented on code in PR #19579:
URL: https://github.com/apache/druid/pull/19579#discussion_r3423219628
##########
processing/src/main/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpec.java:
##########
@@ -0,0 +1,345 @@
+/*
+ * 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.impl;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import com.google.common.collect.Sets;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.error.InvalidInput;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.query.OrderBy;
+import org.apache.druid.query.aggregation.AggregatorFactory;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.column.ColumnHolder;
+import org.apache.druid.utils.CollectionUtils;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * {@link BaseTableProjectionSpec} for the clustered-value-groups base table mode: rows are partitioned into per-tuple
+ * "cluster groups" keyed by one or more typed clustering columns, optionally derived from {@link #virtualColumns}.
+ * Essentially, each group is stored as its own internal sub-segment..
+ * <p>
+ * The operator declares a single ordered {@link #columns} list — the full set of columns in segment order, plus a
+ * {@link #clusteringColumns} list of NAMES designating the leading prefix of {@link #columns} that rows are
+ * clustered by. The time position is an explicit positional entry in {@link #columns} (named {@code __time}, or the
+ * query-granularity column {@link Granularities#GRANULARITY_VIRTUAL_COLUMN_NAME}); clustering by the time column is not
+ * yet supported, so the time marker must be a non-clustering column. A clustered base table is never rollup and has
+ * no metric columns.
+ * <p>
+ * {@link #getDimensionsSpec()} returns the unified spec built from {@link #columns} in declared order with
+ * {@code forceSegmentSortByTime=false}; {@link #getOrdering()} is computed as every column of {@link #columns}
+ * ascending, in list order.
+ * <p>
+ * Query granularity, when wanted, is just another entry in {@link #getVirtualColumns()} named
+ * {@link Granularities#GRANULARITY_VIRTUAL_COLUMN_NAME}; absent that virtual column the query granularity is
+ * {@code NONE}. Segment granularity and intervals live on the top-level
+ * {@link org.apache.druid.indexer.granularity.SegmentGranularitySpec}, not here.
+ */
+@JsonTypeName(ClusteredValueGroupsBaseTableProjectionSpec.TYPE_NAME)
+public final class ClusteredValueGroupsBaseTableProjectionSpec implements BaseTableProjectionSpec
+{
+ public static final String TYPE_NAME = "clusteredValueGroups";
+
+ private final VirtualColumns virtualColumns;
+ private final List<DimensionSchema> columns;
+ private final List<String> clusteringColumns;
+ private final List<DimensionSchema> clusteringColumnSchemas;
+ private final List<DimensionSchema> nonClusteringColumnSchemas;
+ private final DimensionsSpec dimensionsSpec;
+ private final List<OrderBy> ordering;
+
+ public static Builder builder()
+ {
+ return new Builder();
+ }
+
+ @JsonCreator
+ public ClusteredValueGroupsBaseTableProjectionSpec(
+ @JsonProperty("virtualColumns") @Nullable VirtualColumns virtualColumns,
+ @JsonProperty("columns") List<DimensionSchema> columns,
+ @JsonProperty("clusteringColumns") List<String> clusteringColumns
+ )
+ {
+ validate(columns, clusteringColumns);
+ this.virtualColumns = virtualColumns == null ? VirtualColumns.EMPTY : virtualColumns;
+ this.columns = Collections.unmodifiableList(new ArrayList<>(columns));
+ this.clusteringColumns = Collections.unmodifiableList(new ArrayList<>(clusteringColumns));
+ this.clusteringColumnSchemas = this.columns.subList(0, this.clusteringColumns.size());
+ this.nonClusteringColumnSchemas = this.columns.subList(this.clusteringColumns.size(), this.columns.size());
+ this.dimensionsSpec = DimensionsSpec.builder()
+ .setDimensions(this.columns)
+ .setForceSegmentSortByTime(false)
+ .build();
+ this.ordering = computeOrdering(this.columns);
+ }
+
+ @Override
+ @JsonProperty
+ @JsonInclude(JsonInclude.Include.NON_DEFAULT)
+ public VirtualColumns getVirtualColumns()
+ {
+ return virtualColumns;
+ }
+
+ /**
+ * The full, ordered list of columns in segment order; the clustering prefix followed by the remaining
+ * (non-clustering) columns, with the explicit {@code __time} (or query-granularity) marker at its declared
+ * position.
+ */
+ @JsonProperty("columns")
+ public List<DimensionSchema> getColumns()
+ {
+ return columns;
+ }
+
+ /**
+ * The names of the leading prefix of {@link #getColumns()} that rows are clustered by.
+ */
+ @JsonProperty("clusteringColumns")
+ public List<String> getClusteringColumnNames()
+ {
+ return clusteringColumns;
+ }
+
+ /**
+ * The clustering columns as {@link DimensionSchema}s; the leading prefix of {@link #getColumns()}.
+ */
+ @JsonIgnore
+ public List<DimensionSchema> getClusteringColumns()
+ {
+ return clusteringColumnSchemas;
+ }
+
+ /**
+ * The non-clustering columns as {@link DimensionSchema}s; the suffix of {@link #getColumns()} after the
+ * clustering prefix. Includes the explicit time marker.
+ */
+ @JsonIgnore
+ public List<DimensionSchema> getNonClusteringColumns()
+ {
+ return nonClusteringColumnSchemas;
+ }
+
+ @Override
+ public AggregatorFactory[] getMetrics()
+ {
+ return new AggregatorFactory[0];
+ }
+
+ @Override
+ @JsonIgnore
+ public List<OrderBy> getOrdering()
+ {
+ return ordering;
+ }
+
+ @Override
+ @JsonIgnore
+ public DimensionsSpec getDimensionsSpec()
+ {
+ return dimensionsSpec;
+ }
+
+ private static void validate(List<DimensionSchema> columns, List<String> clusteringColumns)
Review Comment:
could we eagerly validate clustering dimension types and throw InvalidInput exception using the DimensionSchema in `columns` for the relevant `clusteringColumns`?
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: clustered segments pt.2 (write side support) (druid)
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis merged PR #19579:
URL: https://github.com/apache/druid/pull/19579
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: clustered segments pt.2 (write side support) (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #19579:
URL: https://github.com/apache/druid/pull/19579#discussion_r3408464125
##########
processing/src/main/java/org/apache/druid/segment/incremental/OnHeapClusterGroup.java:
##########
@@ -0,0 +1,447 @@
+/*
+ * 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.incremental;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.impl.DimensionSchema;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.query.OrderBy;
+import org.apache.druid.query.aggregation.Aggregator;
+import org.apache.druid.query.aggregation.AggregatorAndSize;
+import org.apache.druid.query.aggregation.AggregatorFactory;
+import org.apache.druid.segment.ColumnSelectorFactory;
+import org.apache.druid.segment.EncodedKeyComponent;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.column.CapabilitiesBasedFormat;
+import org.apache.druid.segment.column.ColumnCapabilities;
+import org.apache.druid.segment.column.ColumnCapabilitiesImpl;
+import org.apache.druid.segment.column.ColumnFormat;
+import org.apache.druid.segment.column.ColumnHolder;
+import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.column.ValueType;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * One cluster group within an {@link OnHeapClusteredBaseTable}: a sub-IncrementalIndex holding the non-clustering
+ * rows for a single clustering tuple. The clustering values themselves are NOT stored in this group's facts
+ * holder — they're constants captured on {@link #clusteringValues} (per the share-nothing v1 design, clustering
+ * values are written once on the on-disk group spec rather than per-row).
+ * <p>
+ * Each group owns its own {@link IncrementalIndex.DimensionDesc} list for the non-clustering dimensions, which
+ * means own {@link org.apache.druid.segment.DimensionIndexer} instances and own dictionaries. That isolation is
+ * what gives the persisted segment the "share nothing across groups" property — at persist time each group's
+ * dictionaries are written under its per-group file-bundle prefix.
+ * <p>
+ * Aggregator state is also per-group: rollup (when enabled) deduplicates within a group's facts holder, never
+ * across groups, since two rows with different clustering tuples land in different groups by construction.
+ */
+public final class OnHeapClusterGroup implements IncrementalIndexRowSelector
+{
+ private final OnHeapClusteredBaseTable parent;
+ private final Object[] clusteringValues;
+ private final List<Integer> clusteringValueIds;
+
+ private final List<IncrementalIndex.DimensionDesc> dimensions;
+ private final Map<String, IncrementalIndex.DimensionDesc> dimensionsMap;
+ private final Map<String, IncrementalIndex.MetricDesc> aggregatorsMap;
+ private final Map<String, ColumnFormat> columnFormats;
+ private final AggregatorFactory[] aggregatorFactories;
+ private final FactsHolder factsHolder;
+ private final ConcurrentHashMap<Integer, Aggregator[]> aggregators = new ConcurrentHashMap<>();
+ private final AtomicInteger rowCounter = new AtomicInteger(0);
+ private final AtomicInteger numEntries = new AtomicInteger(0);
+ // Group-local time position within {@link #getOrdering()}: 0 for the canonical "[__time]" group ordering that
+ // results from stripping the clustering-column prefix off the segment ordering.
+ private final int groupTimePosition;
+
+ private final ColumnSelectorFactory virtualSelectorFactory;
+ private final Map<String, ColumnSelectorFactory> aggSelectors;
+
+ OnHeapClusterGroup(
+ OnHeapClusteredBaseTable parent,
+ Object[] clusteringValues,
+ List<Integer> clusteringValueIds,
+ List<DimensionSchema> nonClusteringDimensions,
+ AggregatorFactory[] aggregatorFactories,
+ VirtualColumns virtualColumns,
+ IncrementalIndex.InputRowHolder inputRowHolder,
+ boolean rollup,
+ int timePosition
+ )
+ {
+ this.parent = parent;
+ this.clusteringValues = clusteringValues;
+ this.clusteringValueIds = Collections.unmodifiableList(new ArrayList<>(clusteringValueIds));
+ this.aggregatorFactories = aggregatorFactories;
+
+ this.dimensions = new ArrayList<>(nonClusteringDimensions.size());
+ this.dimensionsMap = new LinkedHashMap<>();
+ this.columnFormats = new LinkedHashMap<>();
+ initializeDimensions(nonClusteringDimensions);
+
+ // Group-local time position. {@code timePosition} arrives from the parent as the SEGMENT-level position of
+ // __time within the segment's full ordering. Cluster groups strip the clustering-column prefix off that
+ // ordering, so the per-group position is shifted by the size of the clustering tuple. For the canonical v1
+ // ordering [clustering..., __time] this comes out to 0.
+ final int clusteringCount = clusteringValues.length;
+ this.groupTimePosition = timePosition < 0 ? -1 : Math.max(0, timePosition - clusteringCount);
+ final int comparatorTimePosition = groupTimePosition < 0 ? dimensions.size() : groupTimePosition;
+
+ final IncrementalIndex.IncrementalIndexRowComparator rowComparator =
+ new IncrementalIndex.IncrementalIndexRowComparator(comparatorTimePosition, dimensions);
+ if (rollup) {
+ this.factsHolder = new OnheapIncrementalIndex.RollupFactsHolder(
+ rowComparator,
+ dimensions,
+ comparatorTimePosition == 0
+ );
+ } else if (comparatorTimePosition == 0) {
+ this.factsHolder = new OnheapIncrementalIndex.PlainTimeOrderedFactsHolder(rowComparator);
+ } else {
+ this.factsHolder = new OnheapIncrementalIndex.PlainNonTimeOrderedFactsHolder(rowComparator);
+ }
+
+ this.virtualSelectorFactory = new OnheapIncrementalIndex.CachingColumnSelectorFactory(
+ IncrementalIndex.makeColumnSelectorFactory(virtualColumns, inputRowHolder, null)
+ );
+ this.aggregatorsMap = new LinkedHashMap<>();
+ this.aggSelectors = new LinkedHashMap<>();
+ initializeAggregators(virtualColumns, inputRowHolder);
+ }
+
+ /**
+ * The constant clustering tuple for this group, in clustering-column declaration order.
+ */
+ public Object[] getClusteringValues()
+ {
+ return clusteringValues;
+ }
+
+ /**
+ * Per-clustering-column dictionary IDs assigned by the parent's segment-wide clustering indexers — these
+ * become {@code TableClusterGroupSpec.clusteringValueIds} at persist time.
+ */
+ public List<Integer> getClusteringValueIds()
+ {
+ return clusteringValueIds;
+ }
+
+ @Override
+ public List<IncrementalIndex.DimensionDesc> getDimensions()
+ {
+ return dimensions;
+ }
+
+ @Override
+ public List<String> getDimensionNames(boolean includeTime)
+ {
+ if (!includeTime) {
+ return ImmutableList.copyOf(dimensionsMap.keySet());
+ }
+ final ImmutableList.Builder<String> listBuilder = ImmutableList.builderWithExpectedSize(dimensionsMap.size() + 1);
+ int i = 0;
+ if (i == groupTimePosition) {
+ listBuilder.add(ColumnHolder.TIME_COLUMN_NAME);
+ }
+ for (String dimName : dimensionsMap.keySet()) {
+ listBuilder.add(dimName);
+ i++;
+ if (i == groupTimePosition) {
+ listBuilder.add(ColumnHolder.TIME_COLUMN_NAME);
+ }
+ }
+ return listBuilder.build();
+ }
+
+ @Override
+ public List<String> getMetricNames()
+ {
+ return ImmutableList.copyOf(aggregatorsMap.keySet());
+ }
+
+ @Override
+ @Nullable
+ public IncrementalIndex.DimensionDesc getDimension(String columnName)
+ {
+ return dimensionsMap.get(columnName);
+ }
+
+ @Override
+ @Nullable
+ public IncrementalIndex.MetricDesc getMetric(String columnName)
+ {
+ return aggregatorsMap.get(columnName);
+ }
+
+ @Override
+ @Nullable
+ public ColumnFormat getColumnFormat(String columnName)
+ {
+ if (ColumnHolder.TIME_COLUMN_NAME.equals(columnName)) {
+ return new CapabilitiesBasedFormat(ColumnCapabilitiesImpl.createDefault().setType(ColumnType.LONG));
+ }
+ return columnFormats.get(columnName);
+ }
+
+ @Override
+ public List<OrderBy> getOrdering()
+ {
+ // Per-group ordering = segment ordering with the clustering-column prefix dropped (clustering values are
+ // constants per group, so they don't participate in intra-group sorting).
+ final List<OrderBy> segmentOrdering = parent.getSpec().getOrdering();
+ final int clusteringCount = clusteringValues.length;
+ if (segmentOrdering.size() <= clusteringCount) {
+ return List.of();
+ }
+ return List.copyOf(segmentOrdering.subList(clusteringCount, segmentOrdering.size()));
+ }
+
+ @Override
+ public int getTimePosition()
+ {
+ return groupTimePosition;
+ }
+
+ @Override
+ public boolean isEmpty()
+ {
+ return numEntries.get() == 0;
+ }
+
+ @Override
+ public int numRows()
+ {
+ return numEntries.get();
+ }
+
+ @Override
+ public FactsHolder getFacts()
+ {
+ return factsHolder;
+ }
+
+ @Override
+ public int getLastRowIndex()
+ {
+ // rowCounter is post-incremented only when a new row is added (see addToFacts), so it holds the count of rows;
+ // the last assigned 0-based index is one less. Mirrors OnheapIncrementalIndex.getLastRowIndex(). Returning the
+ // count instead would let a row appended after a cursor captured this value leak into that in-flight query.
+ return rowCounter.get() - 1;
+ }
+
+ @Override
+ public float getMetricFloatValue(int rowOffset, int aggOffset)
+ {
+ return aggregators.get(rowOffset)[aggOffset].getFloat();
+ }
+
+ @Override
+ public long getMetricLongValue(int rowOffset, int aggOffset)
+ {
+ return aggregators.get(rowOffset)[aggOffset].getLong();
+ }
+
+ @Override
+ public double getMetricDoubleValue(int rowOffset, int aggOffset)
+ {
+ return aggregators.get(rowOffset)[aggOffset].getDouble();
+ }
+
+ @Override
+ @Nullable
+ public Object getMetricObjectValue(int rowOffset, int aggOffset)
+ {
+ return aggregators.get(rowOffset)[aggOffset].get();
+ }
+
+ @Override
+ public boolean isNull(int rowOffset, int aggOffset)
+ {
+ return aggregators.get(rowOffset)[aggOffset].isNull();
+ }
+
+ @Override
+ @Nullable
+ public ColumnCapabilities getColumnCapabilities(String column)
+ {
+ if (ColumnHolder.TIME_COLUMN_NAME.equals(column)) {
+ return ColumnCapabilitiesImpl.createDefault().setType(ColumnType.LONG).setHasNulls(false);
+ }
+ final IncrementalIndex.DimensionDesc dim = dimensionsMap.get(column);
+ if (dim != null) {
+ return dim.getCapabilities();
+ }
+ final IncrementalIndex.MetricDesc metric = aggregatorsMap.get(column);
+ if (metric != null) {
+ return metric.getCapabilities();
+ }
+ return null;
+ }
+
+ public AggregatorFactory[] getAggregatorFactories()
+ {
+ return aggregatorFactories;
+ }
+
+ /**
+ * Add a row's non-clustering content to this group's facts holder. Clustering values aren't re-processed here —
+ * they've already been resolved + encoded by the parent {@link OnHeapClusteredBaseTable} as part of selecting
+ * this group. Returns true when the row created a new fact entry (vs. rolling up into an existing one), so the
+ * parent index can keep its segment-wide entry count in sync.
+ */
+ boolean addToFacts(
+ InputRow row,
+ long bucketedTimestamp,
+ List<String> parseExceptionMessages,
+ AtomicLong totalSizeInBytes
+ )
+ {
+ final Object[] groupDims = new Object[dimensions.size()];
+ long dimsKeySize = 0L;
+ for (int i = 0; i < dimensions.size(); i++) {
+ final IncrementalIndex.DimensionDesc desc = dimensions.get(i);
+ try {
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ final EncodedKeyComponent<?> k = ((org.apache.druid.segment.DimensionIndexer) desc.getIndexer())
+ .processRowValsToUnsortedEncodedKeyComponent(row.getRaw(desc.getName()), true);
+ groupDims[i] = k.getComponent();
+ dimsKeySize += k.getEffectiveSizeBytes();
+ }
+ catch (org.apache.druid.java.util.common.parsers.ParseException pe) {
+ parseExceptionMessages.add(pe.getMessage());
+ }
+ }
+ totalSizeInBytes.addAndGet(dimsKeySize);
+
+ final IncrementalIndexRow subKey = IncrementalIndexRow.createTimeAndDimswithDimsKeySize(
+ bucketedTimestamp,
+ groupDims,
+ dimensions,
+ dimsKeySize
+ );
+
+ final int priorIndex = factsHolder.getPriorIndex(subKey);
+ final Aggregator[] aggs;
+ if (IncrementalIndexRow.EMPTY_ROW_INDEX != priorIndex) {
+ aggs = aggregators.get(priorIndex);
+ final long aggSizeDelta = OnheapIncrementalIndex.doAggregate(
+ aggregatorFactories,
+ aggs,
+ parent.getInputRowHolder(),
+ parseExceptionMessages,
+ false
+ );
+ totalSizeInBytes.addAndGet(aggSizeDelta);
+ return false;
+ } else {
+ aggs = new Aggregator[aggregatorFactories.length];
+ long aggSizeForRow = factorizeAggs(aggs);
+ aggSizeForRow += OnheapIncrementalIndex.doAggregate(
+ aggregatorFactories,
+ aggs,
+ parent.getInputRowHolder(),
+ parseExceptionMessages,
+ false
+ );
+ final int rowIndex = rowCounter.getAndIncrement();
+ aggregators.put(rowIndex, aggs);
+ final int prev = factsHolder.putIfAbsent(subKey, rowIndex);
+ if (IncrementalIndexRow.EMPTY_ROW_INDEX == prev) {
+ numEntries.incrementAndGet();
+ } else {
+ throw DruidException.defensive("Encountered existing fact entry for new key in cluster group");
+ }
+ final long rowSize = subKey.estimateBytesInMemory()
+ + aggSizeForRow
+ + OnheapIncrementalIndex.ROUGH_OVERHEAD_PER_MAP_ENTRY;
+ totalSizeInBytes.addAndGet(rowSize);
+ return true;
+ }
+ }
+
+ private void initializeDimensions(List<DimensionSchema> nonClusteringDimensions)
+ {
+ int i = 0;
+ for (DimensionSchema schema : nonClusteringDimensions) {
+ if (ColumnHolder.TIME_COLUMN_NAME.equals(schema.getName())) {
+ // __time is handled via the row timestamp, not as a dim desc
+ continue;
+ }
+ final IncrementalIndex.DimensionDesc desc = new IncrementalIndex.DimensionDesc(
+ i++,
+ schema.getName(),
+ schema.getDimensionHandler()
+ );
+ dimensions.add(desc);
+ dimensionsMap.put(schema.getName(), desc);
+ columnFormats.put(schema.getName(), desc.getIndexer().getFormat());
+ }
+ }
+
+ private void initializeAggregators(
+ VirtualColumns virtualColumns,
Review Comment:
## CodeQL / Useless parameter
The parameter 'virtualColumns' is never used.
[Show more details](https://github.com/apache/druid/security/code-scanning/11308)
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: clustered segments pt.2 (write side support) (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #19579:
URL: https://github.com/apache/druid/pull/19579#discussion_r3420357309
##########
processing/src/main/java/org/apache/druid/segment/incremental/OnHeapClusteredBaseTable.java:
##########
@@ -0,0 +1,491 @@
+/*
+ * 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.incremental;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Ordering;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.impl.ClusteredValueGroupsBaseTableProjectionSpec;
+import org.apache.druid.data.input.impl.DimensionSchema;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.java.util.common.parsers.ParseException;
+import org.apache.druid.segment.ColumnSelectorFactory;
+import org.apache.druid.segment.ColumnValueSelector;
+import org.apache.druid.segment.DimensionDictionary;
+import org.apache.druid.segment.DimensionHandlerUtils;
+import org.apache.druid.segment.SortedDimensionDictionary;
+import org.apache.druid.segment.StringDimensionDictionary;
+import org.apache.druid.segment.VirtualColumn;
+import org.apache.druid.segment.VirtualColumns;
+import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.segment.column.ValueType;
+import org.apache.druid.segment.projections.ClusteredValueGroupsBaseTableSchema;
+import org.apache.druid.segment.projections.ClusteringDictionaries;
+import org.apache.druid.segment.projections.TableClusterGroupSpec;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Segment-wide root of the clustered base-table machinery for {@link OnheapIncrementalIndex}: holds the per-type
+ * shared clustering dictionaries, the clustering virtual-column selector factory, and the map of
+ * {@link OnHeapClusterGroup} instances keyed by the clustering-value dictionary-ID tuple. The parent
+ * {@code OnheapIncrementalIndex} delegates row ingestion here when its schema carries a
+ * {@link ClusteredValueGroupsBaseTableProjectionSpec}, and the base facts holder stays empty in that mode.
+ * <p>
+ * On row arrival: each clustering column's value is resolved through the virtual-column selector, then
+ * dictionary-encoded into the per-type dictionary for the column's clustering type. The resulting dictionary-ID tuple
+ * becomes the lookup key for the {@link OnHeapClusterGroup} that owns the row, new groups are materialized lazily as
+ * new tuples arrive.
+ * <p>
+ * Dictionaries here are insertion-order (id = first-seen position). The persist path sorts + remaps them into the
+ * read-side {@link ClusteringDictionaries} shape (sorted, nulls first) at segment-write time.
+ */
+public final class OnHeapClusteredBaseTable
+{
+ private static final Comparator<TableClusterGroupSpec> BY_CLUSTERING_VALUE_IDS =
+ Comparator.comparing(TableClusterGroupSpec::getClusteringValueIds, Ordering.<Integer>natural().lexicographical());
+
+ private final ClusteredValueGroupsBaseTableProjectionSpec spec;
+ private final RowSignature clusteringColumns;
+ private final VirtualColumns virtualColumns;
+ private final ColumnSelectorFactory virtualSelectorFactory;
+ private final IncrementalIndex.InputRowHolder inputRowHolder;
+
+ // template state for instantiating new OnHeapClusterGroups
+ private final List<DimensionSchema> nonClusteringDimensions;
+ private final boolean rollup;
+ private final int timePosition;
+
+ private final DimensionDictionary<String> stringDictionary = new StringDimensionDictionary();
+ private final DimensionDictionary<Long> longDictionary = new DimensionDictionary<>(Long.class)
+ {
+ @Override
+ public long estimateSizeOfValue(Long value)
+ {
+ return Long.BYTES;
+ }
+ };
+ private final DimensionDictionary<Double> doubleDictionary = new DimensionDictionary<>(Double.class)
+ {
+ @Override
+ public long estimateSizeOfValue(Double value)
+ {
+ return Double.BYTES;
+ }
+ };
+ private final DimensionDictionary<Float> floatDictionary = new DimensionDictionary<>(Float.class)
+ {
+ @Override
+ public long estimateSizeOfValue(Float value)
+ {
+ return Float.BYTES;
+ }
+ };
+
+ // Keyed by the clustering-value dictionary-ID tuple
+ private final ConcurrentHashMap<List<Integer>, OnHeapClusterGroup> groups = new ConcurrentHashMap<>();
+ private final AtomicInteger totalNumRows = new AtomicInteger(0);
+ // min/max bucketed row timestamp across all groups; the base facts holder is empty in clustered mode so the
+ // parent index's interval accessors delegate here instead.
+ private final AtomicLong minTimeMillis = new AtomicLong(Long.MAX_VALUE);
+ private final AtomicLong maxTimeMillis = new AtomicLong(Long.MIN_VALUE);
+
+ public OnHeapClusteredBaseTable(
+ ClusteredValueGroupsBaseTableProjectionSpec spec,
+ VirtualColumns segmentVirtualColumns,
+ IncrementalIndex.InputRowHolder inputRowHolder,
+ boolean rollup,
+ int timePosition
+ )
+ {
+ this.spec = spec;
+ this.inputRowHolder = inputRowHolder;
+ this.rollup = rollup;
+ this.timePosition = timePosition;
+ this.nonClusteringDimensions = Collections.unmodifiableList(
+ new ArrayList<>(spec.getNonClusteringColumns())
+ );
+
+ final RowSignature.Builder sigBuilder = RowSignature.builder();
+ for (DimensionSchema c : spec.getClusteringColumns()) {
+ sigBuilder.add(c.getName(), c.getColumnType());
+ }
+ this.clusteringColumns = sigBuilder.build();
+
+ this.virtualColumns = mergeVirtualColumns(segmentVirtualColumns, spec.getVirtualColumns());
+ this.virtualSelectorFactory = new OnheapIncrementalIndex.CachingColumnSelectorFactory(
+ IncrementalIndex.makeColumnSelectorFactory(this.virtualColumns, inputRowHolder, null)
+ );
+ }
+
+ /**
+ * Resolve the clustering tuple for {@code row}, locate (or create) the matching {@link OnHeapClusterGroup}, and
+ * dispatch the row to it. {@code key} carries the bucketed timestamp from the parent's {@code toIncrementalIndexRow}
+ * pre-processing, its dim slots are ignored here since in clustered mode the non-clustering data lives only on the
+ * per-group facts holders. Returns true when the row created a new fact entry in its group (vs. rolling up into an
+ * existing one).
+ */
+ boolean addToFacts(
+ IncrementalIndexRow key,
+ InputRow row,
+ List<String> parseExceptionMessages,
+ AtomicLong totalSizeInBytes
+ )
+ {
+ final long clusteringDictSizeBefore = clusteringDictionariesSizeInBytes();
+ final Object[] clusteringValues = new Object[clusteringColumns.size()];
+ final List<Integer> clusteringValueIds = new ArrayList<>(clusteringColumns.size());
+ for (int i = 0; i < clusteringColumns.size(); i++) {
+ final String name = clusteringColumns.getColumnName(i);
+ final ColumnType type = clusteringColumns.getColumnType(i)
+ .orElseThrow(() -> DruidException.defensive(
+ "clustering column [%s] has no type",
+ name
+ ));
+ final ColumnValueSelector<?> selector = virtualSelectorFactory.makeColumnValueSelector(name);
+ final Object raw = selector.getObject();
+ Object coerced;
+ try {
+ coerced = DimensionHandlerUtils.convertObjectToType(raw, type, true, name);
+ }
+ catch (ParseException pe) {
+ parseExceptionMessages.add(pe.getMessage());
+ coerced = null;
+ }
+ clusteringValues[i] = coerced;
+ clusteringValueIds.add(internClusteringValue(type, coerced));
+ }
+ totalSizeInBytes.addAndGet(clusteringDictionariesSizeInBytes() - clusteringDictSizeBefore);
+
+ final boolean[] groupCreated = {false};
+ final OnHeapClusterGroup group = groups.computeIfAbsent(
+ clusteringValueIds,
+ ids -> {
+ groupCreated[0] = true;
+ return new OnHeapClusterGroup(
+ this,
+ clusteringValues,
+ ids,
+ nonClusteringDimensions,
+ virtualColumns,
+ inputRowHolder,
+ rollup,
+ timePosition
+ );
+ }
+ );
+ if (groupCreated[0]) {
+ totalSizeInBytes.addAndGet(estimateNewGroupOverhead());
+ }
+ final boolean isNewEntry = group.addToFacts(row, key.getTimestamp(), parseExceptionMessages, totalSizeInBytes);
+ totalNumRows.incrementAndGet();
+ minTimeMillis.accumulateAndGet(key.getTimestamp(), Math::min);
+ maxTimeMillis.accumulateAndGet(key.getTimestamp(), Math::max);
+ return isNewEntry;
+ }
+
+ /**
+ * Rough running-heap cost of a newly-created cluster group beyond its rows: the {@link #groups}-map node, the boxed
+ * clustering-id tuple held as the map key, the per-group clustering-values array, and the group object's fixed
+ * structures.
+ */
+ private long estimateNewGroupOverhead()
+ {
+ final int numClusteringColumns = clusteringColumns.size();
+ return OnheapIncrementalIndex.ROUGH_OVERHEAD_PER_MAP_ENTRY
+ + (long) numClusteringColumns * (Long.BYTES * 2 + Long.BYTES)
Review Comment:
## CodeQL / Result of multiplication cast to wider type
Potential overflow in [int multiplication](1) before it is converted to long by use in a numeric context.
[Show more details](https://github.com/apache/druid/security/code-scanning/11309)
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] feat: clustered segments pt.2 (write side support) (druid)
Posted by "capistrant (via GitHub)" <gi...@apache.org>.
capistrant commented on code in PR #19579:
URL: https://github.com/apache/druid/pull/19579#discussion_r3424171847
##########
processing/src/main/java/org/apache/druid/data/input/impl/ClusteredValueGroupsBaseTableProjectionSpec.java:
##########
@@ -193,40 +206,45 @@ private static void validate(List<DimensionSchema> columns, List<String> cluster
}
int timeIndex = -1;
- boolean bothPresent = false;
for (int i = 0; i < columns.size(); i++) {
final String name = columns.get(i).getName();
- if (ColumnHolder.TIME_COLUMN_NAME.equals(name) || Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME.equals(name)) {
- if (timeIndex >= 0) {
- bothPresent = true;
- }
+ // The query-granularity virtual column is a granularity carrier in virtualColumns (it floors the stored __time
+ // column); it is not itself a stored column, so it must not be declared in 'columns'.
+ if (Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME.equals(name)) {
+ throw InvalidInput.exception(
+ "[%s] is the query-granularity virtual column, not a stored column; declare it in 'virtualColumns' and use"
+ + " [%s] as the time column in 'columns'",
+ Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME,
+ ColumnHolder.TIME_COLUMN_NAME
+ );
+ }
+ if (ColumnHolder.TIME_COLUMN_NAME.equals(name)) {
timeIndex = i;
}
}
if (timeIndex < 0) {
throw InvalidInput.exception(
- "clustered base table must include %s (or the query-granularity column [%s]) in 'columns' to define the"
- + " time position",
- ColumnHolder.TIME_COLUMN_NAME,
- Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME
- );
- }
- if (bothPresent) {
- throw InvalidInput.exception(
- "clustered base table must include exactly one of %s / %s in 'columns' to define the time position",
- ColumnHolder.TIME_COLUMN_NAME,
- Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME
+ "clustered base table must include [%s] in 'columns' to define the time position",
+ ColumnHolder.TIME_COLUMN_NAME
);
}
if (timeIndex < clusteringColumns.size()) {
throw InvalidInput.exception(
- "clustering by %s / %s is not yet supported; the time column must be a non-clustering column",
- ColumnHolder.TIME_COLUMN_NAME,
- Granularities.GRANULARITY_VIRTUAL_COLUMN_NAME
+ "clustering by [%s] is not yet supported; the time column must be a non-clustering column",
+ ColumnHolder.TIME_COLUMN_NAME
);
}
}
+ private static boolean isSupportedClusteringType(ColumnType type)
Review Comment:
`Projections.isAllowedClusteringType` is an existing public utility doing the same. I think consolidating would be helpful to make future changes to supported clustering types (if any) require less fixups
--
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: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org