Segments sorted by non-time columns.

[gianm]

Summary

Currently, segments are always sorted by __time, followed by the sort order provided by the user via dimensionsSpec or CLUSTERED BY. Sorting by __time enables efficient execution of queries involving time-ordering or granularity. Time-ordering is a simple matter of reading the rows in stored order, and granular cursors can be generated in streaming fashion.

However, for various workloads, it's better for storage footprint and query performance to sort by arbitrary orders that do not start with __time. With this patch, users can sort segments by such orders.

API

For spec-based ingestion, users add forceSegmentSortByTime: false to dimensionsSpec. The dimensions list determines the sort order. To define a sort order that includes __time, users explicitly include a dimension named __time.

For SQL-based ingestion, users set the context parameter forceSegmentSortByTime: false. The CLUSTERED BY clause is then used as the explicit segment sort order.

In both cases, when the new forceSegmentSortByTime parameter is true (the default), __time is implicitly prepended to the sort order, as it always was prior to this patch.

The new parameter is experimental for two main reasons. First, such segments can cause errors when loaded by older servers, due to violating their expectations that timestamps are always monotonically increasing. Second, even on newer servers, not all queries can run on non-time-sorted segments. Scan queries involving time-ordering and any query involving granularity will not run. (To partially mitigate this, a currently-undocumented SQL feature sqlUseGranularity is provided. When set to false the SQL planner avoids using granularity.)

Main changes

Changes on the write path:

1. DimensionsSpec can now optionally contain a __time dimension, which controls the placement of __time in the sort order. If not present, __time is considered to be first in the sort order, as it has always been.

2. IncrementalIndex and IndexMerger are updated to sort facts more flexibly; not always by time first.

3. Metadata (stored in metadata.drd) gains a sortOrder field.

4. MSQ can generate range-based shard specs even when not all columns are singly-valued strings. It merely stops accepting new clustering key fields when it encounters the first one that isn't a singly-valued string. This is useful because it enables range shard specs on someDim to be created for clauses like CLUSTERED BY someDim, __time.

5. Auto-compaction respects and propagates sort orders that don't start with __time.

Changes on the read path:

1. Update cursor holders for QueryableIndex and IncrementalIndex to return the ordering of the underlying index, so query engines can tell how a segment is sorted.

2. Update CursorGranularizer and VectorCursorGranularizer to throw errors when using granularities on non-time-ordered segments.

3. Update ScanQueryEngine to throw an error when using the time-ordering order parameter on non-time-ordered segments.

4. Update TimeBoundaryQueryRunnerFactory to perform a segment scan when running on a non-time-ordered segment.

5. Add sqlUseGranularity context parameter that causes the SQL planner to avoid using granularities other than ALL. This is undocumented, because it's hopefully a short-term hack. The more ideal thing would be to have all the native queries work properly on these segments.

6. Move getMinTime, getMaxTime, and getMaxIngestedEventTime from StorageAdapter to TimeBoundaryInspector and MaxIngestedEventTimeInspector. This is mainly necessary for timeBoundary queries to be able to tell whether they can use getMinTime and getMaxTime, vs. needing to use a cursor. Previously, timeBoundary assumed that an adapter backing a TableDataSource was guaranteed to have an exact getMinTime and getMaxTime. But after this patch, that isn't necessarily going to be the case (in particular: a non-time-sorted segment won't be able to know its exact min/max time without a full scan.)

Other changes:

1. Rename DimensionsSpec#hasCustomDimensions to hasFixedDimensions and change the meaning subtly: it now returns true if the DimensionsSpec represents an unchanging list of dimensions, or false if there is some discovery happening. This is what call sites had expected anyway.

2. Removed descending from Joinable#makeJoinMatcher. Now that descending has more or less been phased out in favor of ordering: [__time DESC], the joinable order no longer needs to be reversed. (All joined rows arising from the same left-hand row have the same value.)

[clintropolis]

I know we need this right now, but i'm not sure we will need this after #16533. timeRangeIterable is primarily used to support query granularity buckets in topN (in my branch to support mark/resetToMark to move the cursor in the facts table to the correct granularity bucket without having to advance the cursor directly). i think we could just use iterator or expose an alternative iteralble for incremental index cursor if we aren't requesting time ordering

[gianm]

I think it's also used for intervals filtering even for non-granular cursors. It seems like it'd be useful for that at least.

[clintropolis]

descending is pretty tightly coupled with time ordering, but i guess is harmless to implement

[gianm]

It seemed to make more sense to implement it here rather than throw an error. I suppose in theory there could be a use case, in the future, for situations like ORDER BY userId DESC when the segment is sorted by userId.

[clintropolis]

given that we will basically always be reading the unnested column, i guess the main expected utility of this will be if ordering by one of the columns that is not being unnested?

[gianm]

I figure the common case is that the unnested column will not be first in the sort order. So we'll return whatever prefix of the order is there up to the unnested column.

[clintropolis]

in the context of #16533, this should probably live on CursorHolder instead of StorageAdapter, though it will require a bunch of tests to make and dispose of a cursor holder to check if their query against the sort order, but otherwise shouldn't be very disruptive. I wonder if we should include a direction similar to in that PR though, maybe re-using scan query ordering so that these two changes are compatible?

[gianm]

This sounds like a good idea, although in the interests of minimizing conflicts, it'd probably be good to do this after #16533 is merged (since it moves OrderBy around).

[clintropolis]

i know it cannot be specified during ingest with the current mechanisms, but it seems like we should include the direction of the ordering as well, maybe re-use the scan query ordering?

[gianm]

see #16849 (comment)

[clintropolis]

should this be an error if they aren't all null? seems like it should be pretty consistent across indexable adapters...

[gianm]

I made it lenient for two reasons:

• Merging of the other parts of the Metadata is also lenient
• It isn't documented that this method requires that all Metadatas are sourced from segments created with the same ingestion spec + the same version of the software

[clintropolis]

does this happen from columns with all null values or something? maybe this method could use some comments to make the rationale behind the decisions clearer

[gianm]

I added some comments and also fixed a problem: null is now treated as [__time] (which makes sense, since that was the only possible sort order prior to the sortOrder field being added). Due to this change the method is no longer @Nullable.

[clintropolis]

is this referring to the RowWalker with its skipToDateTime?

[gianm]

Yes.

[clintropolis]

maybe we should always write this out instead of leaving it null? i guess it is done like this to make it easier to fill in older segments and new segments ordered by time first?

[gianm]

Hmm I don't totally remember why this is conditional. It doesn't really make sense to be conditional IMO. I updated it to always write out.

[clintropolis]

i suppose depending on the type of facts holder we could actually report something here (rollup should be ordered i think?)

[gianm]

I think that makes sense. I added a comment about this being a possible change in the future.

[gianm]

From the discussion in #16849 (comment), after #16533 is merged we should revisit this one to (a) resolve conflict (b) replace List<String> with List<OrderBy> in the sort order field.

[gianm]

Pushed up a commit resolving conflicts with #16533. Main changes:

• Changed frame cursors to report "no ordering" rather than time ordering.
• Removed descending from Joinable#makeJoinMatcher. Now that descending has more or less been phased out in favor of ordering: [__time DESC], the joinable order no longer needs to be reversed. (All joined rows arising from the same left-hand row have the same value.)
• Moved granularity validations (the ones that make the message Cannot use granularity[%s] on non-time-sorted data) into the granularizers.
• Moved getMinTime, getMaxTime, and getMaxIngestedEventTime from StorageAdapter to TimeBoundaryInspector and MaxIngestedEventTimeInspector. This is mainly necessary for timeBoundary queries to be able to tell whether they can use getMinTime and getMaxTime, vs. needing to use a cursor. Previously, timeBoundary assumed that an adapter backing a TableDataSource was guaranteed to have an exact getMinTime and getMaxTime. But after this patch, that isn't necessarily going to be the case (in particular: a non-time-sorted segment won't be able to know its exact min/max time without a full scan.)

[maytasm]

@clintropolis @gianm
I think this PR broke dataSourceMetadata query.
Seems like MaxIngestedEventTimeInspector in DataSourceMetadataQueryRunner will always be null.
This can be reproduced with the wikipedia dataset.
Just run

{
"queryType" : "dataSourceMetadata",
"dataSource" : "wikipedia"
}

The "maxIngestedEventTime" returned is -146136543-09-08T08:23:32.096Z is wrong.

Doesn't look like this code in QueryableIndexSegment:

  @SuppressWarnings("unchecked")
  @Nullable
  @Override
  public <T> T as(@Nonnull Class<T> clazz)
  {
    final Function<QueryableIndexSegment, ?> fn = AS_MAP.get(clazz);
    if (fn != null) {
      final T fnApply = (T) fn.apply(this);
      if (fnApply != null) {
        return fnApply;
      }
    }

    if (TimeBoundaryInspector.class.equals(clazz)) {
      return (T) timeBoundaryInspector;
    } else if (Metadata.class.equals(clazz)) {
      return (T) index.getMetadata();
    } else if (PhysicalSegmentInspector.class.equals(clazz)) {
      return (T) new QueryableIndexPhysicalSegmentInspector(index);
    } else if (TopNOptimizationInspector.class.equals(clazz)) {
      return (T) new SimpleTopNOptimizationInspector(true);
    }

    return Segment.super.as(clazz);
  }

Is handling MaxIngestedEventTimeInspector?

[clintropolis]

I think this PR broke dataSourceMetadata query.
Seems like MaxIngestedEventTimeInspector in DataSourceMetadataQueryRunner will always be null.

The intention of the new MaxIngestedEventTimeInspector is that it is only for use with realtime data, because every other implementation on the old StorageAdapter of getMaxIngestedEventTime() was just calling getMaxTime() which gives the end interval of the segment. Could use timeBoundary instead if there is no realtime data.

It does seem odd we return min datetime, i would have expected null. I don't think it makes much sense to have an implementation of MaxIngestedEventTimeInspector for published segments given that timeBoundary exists, but the docs could probably more clearly spell this out. Apologies for the behavior change.

[maytasm]

@clintropolis
Thanks for the response.

• Im a little confused on when you said that getMaxTime() gives the end interval of the segment. I believe that this query (dataSourceMetadata) was previously returning something like 2025-01-21T08:23:32.096Z for our non-realtime datasource. The code for getMaxTime() in QueryableIndexStorageAdapter looks something like:

  private void populateMinMaxTime()
  {
    // Compute and cache minTime, maxTime.
    final ColumnHolder columnHolder = index.getColumnHolder(ColumnHolder.TIME_COLUMN_NAME);
    try (NumericColumn column = (NumericColumn) columnHolder.getColumn()) {
      this.minTime = DateTimes.utc(column.getLongSingleValueRow(0));
      this.maxTime = DateTimes.utc(column.getLongSingleValueRow(column.length() - 1));
    }
  }

which does seems to return the actual time (min and max) of the rows/data (not the end interval of the segment)... unless I am missing something.

• Given that the intention of the new MaxIngestedEventTimeInspector is for only realtime data, is it still fair to say that the dataSourceMetadata query should still work for non-realtime datasource (as it was previously working for non-realtime datasources)?
• It is returning min datetime as currMaxIngestedEventTime is null at: https://github.com/apache/druid/blob/master/processing/src/main/java/org/apache/druid/query/datasourcemetadata/DataSourceMetadataQuery.java#L109

[maytasm]

• Ah, the timeBoundary does return the actual time (min and max) of the rows/data which matches with the previous dataSourceMetadata. We can switch to the timeBoundary query as a workaround. Thanks!

[clintropolis]

Im a little confused on when you said that getMaxTime() gives the end interval of the segment.

Oh sorry, i meant the max timestamp within the segment interval, that code is the same (ish), its just on TimeBoundaryInspector now, https://github.com/apache/druid/blob/master/processing/src/main/java/org/apache/druid/segment/QueryableIndexTimeBoundaryInspector.java#L85

[gianm]

Oops, the behavior change was unintentional (at least I didn't intend it, and this was my PR). This should restore the old behavior: #17686