Skip to content

Commit db76edf

Browse files
committed
Honour Calcite TIMESTAMP precision in QTF stitcher output schema
ArrowCalciteTypes.toArrow() hardcoded TimeUnit.MILLISECOND for every Calcite TIMESTAMP, so the late-materialization Stitcher pre-allocated a TimestampMilliVector for the output VSR. Shards correctly emitted Timestamp(NANOSECOND) for date_nanos fields, and Arrow's BaseFixedWidthVector.copyFromSafe requires identical fixed-width subclasses, so it tripped on every shard's response with a message-less IllegalArgumentException — surfacing to clients as HTTP 400 "Invalid Query". Plan shape required to fire: multi-shard parquet-backed index with date_nanos + Timestamp in projection + sort + head N (or other top-K) + at least one fetch-only field above the anchor. Drop any one and the rewriter declines QTF or the bug doesn't fire. Fix: branch on Calcite precision. precision == 9 -> NANOSECOND, else MILLISECOND. BackendPlanAdapter preserves precision through CAST rewrites, so the value is reliable (verified via DEBUG plan dump showing TIMESTAMP(9) survives). - Unit: ArrowCalciteTypesTests adds 3 cases covering precision 3, 9, default. The factory uses an extended RelDataTypeSystemImpl that lifts MAX_DATETIME precision to 9 — default Calcite caps at 3 and would silently clamp the test input. - IT: LateMaterializationDateNanosIT spins a 2-shard parquet+lucene index with date_nanos and runs the q01-style plan (match + sort - ts + head + fetch-only field). Pre-fix: HTTP 400 with the exact bug signature. Post-fix: 4 rows in DESC order with sub-millisecond precision preserved. Signed-off-by: Vinay Krishna Pudyodu <vinkrish.neo@gmail.com>
1 parent 3cd77ad commit db76edf

3 files changed

Lines changed: 141 additions & 11 deletions

File tree

sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/ArrowCalciteTypes.java

Lines changed: 2 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -22,12 +22,6 @@
2222
* schema. The {@code AggregateFunction.ArrowToCalciteTypeMapper} (in the SPI module) handles
2323
* the inverse direction for {@code IntermediateField} resolution; this class is kept as the
2424
* single authority for the Calcite→Arrow direction needed outside that resolver.
25-
*
26-
* <p>FIXME [FixBeforeMainMerge] coverage gaps: TIMESTAMP currently hardcodes MILLISECOND
27-
* (Calcite precision is ignored — see toArrow note), so date_nanos is not yet distinguished,
28-
* and several Calcite/Arrow types are still unmapped (TIMESTAMP_WITH_LOCAL_TIME_ZONE, DATE,
29-
* TIME, DECIMAL, Arrow Date/Time/Decimal/...). Audit and broaden before merge so the QTF path
30-
* tolerates non-keyword/non-date columns end-to-end.
3125
*/
3226
public final class ArrowCalciteTypes {
3327

@@ -53,11 +47,8 @@ public static ArrowType toArrow(RelDataType t) {
5347
case VARBINARY, BINARY -> ArrowType.Binary.INSTANCE;
5448
case BOOLEAN -> ArrowType.Bool.INSTANCE;
5549
// TODO: TIMESTAMP_WITH_LOCAL_TIME_ZONE, DATE, TIME, DECIMAL still missing.
56-
// TODO: hardcoded MILLISECOND to match what DateParquetField emits on the data node;
57-
// Calcite's reported precision doesn't track the wire-level Arrow precision today, so
58-
// honouring t.getPrecision() here would break Stitcher copyFromSafe. Revisit when
59-
// Calcite types carry the data-node-side Arrow precision faithfully.
60-
case TIMESTAMP -> new ArrowType.Timestamp(TimeUnit.MILLISECOND, null);
50+
// precision 9 ⇒ date_nanos; else date — must match the wire unit shards emit (Stitcher copyFromSafe).
51+
case TIMESTAMP -> new ArrowType.Timestamp(t.getPrecision() == 9 ? TimeUnit.NANOSECOND : TimeUnit.MILLISECOND, null);
6152
default -> throw new IllegalArgumentException("Unsupported Calcite type: " + t.getSqlTypeName());
6253
};
6354
}

sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/ArrowCalciteTypesTests.java

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,9 +8,11 @@
88

99
package org.opensearch.analytics.planner;
1010

11+
import org.apache.arrow.vector.types.TimeUnit;
1112
import org.apache.arrow.vector.types.pojo.ArrowType;
1213
import org.apache.calcite.rel.type.RelDataType;
1314
import org.apache.calcite.rel.type.RelDataTypeSystem;
15+
import org.apache.calcite.rel.type.RelDataTypeSystemImpl;
1416
import org.apache.calcite.sql.type.SqlTypeFactoryImpl;
1517
import org.apache.calcite.sql.type.SqlTypeName;
1618
import org.opensearch.test.OpenSearchTestCase;
@@ -23,10 +25,25 @@ public class ArrowCalciteTypesTests extends OpenSearchTestCase {
2325

2426
private static final SqlTypeFactoryImpl TYPE_FACTORY = new SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT);
2527

28+
/** Lifts TIMESTAMP max-precision to 9; default Calcite caps at 3 and would clamp date_nanos away. */
29+
private static final SqlTypeFactoryImpl NANOS_TYPE_FACTORY = new SqlTypeFactoryImpl(new RelDataTypeSystemImpl() {
30+
@Override
31+
public int getMaxPrecision(SqlTypeName typeName) {
32+
if (typeName == SqlTypeName.TIMESTAMP || typeName == SqlTypeName.TIMESTAMP_WITH_LOCAL_TIME_ZONE) {
33+
return 9;
34+
}
35+
return super.getMaxPrecision(typeName);
36+
}
37+
});
38+
2639
private static RelDataType type(SqlTypeName name) {
2740
return TYPE_FACTORY.createSqlType(name);
2841
}
2942

43+
private static RelDataType timestamp(int precision) {
44+
return NANOS_TYPE_FACTORY.createSqlType(SqlTypeName.TIMESTAMP, precision);
45+
}
46+
3047
/**
3148
* SMALLINT (OpenSearch {@code short}) must map to the wire Arrow type the data node
3249
* emits — {@code Int(16, true)} per {@code ShortParquetField} — so the Stitcher's
@@ -45,4 +62,19 @@ public void testIntegerAndBigintUnchanged() {
4562
assertEquals(new ArrowType.Int(32, true), ArrowCalciteTypes.toArrow(type(SqlTypeName.INTEGER)));
4663
assertEquals(new ArrowType.Int(64, true), ArrowCalciteTypes.toArrow(type(SqlTypeName.BIGINT)));
4764
}
65+
66+
/** date ⇒ TIMESTAMP(3) ⇒ MILLISECOND. */
67+
public void testTimestampPrecision3MapsToMillisecond() {
68+
assertEquals(new ArrowType.Timestamp(TimeUnit.MILLISECOND, null), ArrowCalciteTypes.toArrow(timestamp(3)));
69+
}
70+
71+
/** date_nanos ⇒ TIMESTAMP(9) ⇒ NANOSECOND — regression: previously hardcoded MILLISECOND, tripped Stitcher copyFromSafe. */
72+
public void testTimestampPrecision9MapsToNanosecond() {
73+
assertEquals(new ArrowType.Timestamp(TimeUnit.NANOSECOND, null), ArrowCalciteTypes.toArrow(timestamp(9)));
74+
}
75+
76+
/** Default-precision TIMESTAMP (precision 0) keeps the legacy MILLISECOND mapping. */
77+
public void testTimestampDefaultPrecisionMapsToMillisecond() {
78+
assertEquals(new ArrowType.Timestamp(TimeUnit.MILLISECOND, null), ArrowCalciteTypes.toArrow(type(SqlTypeName.TIMESTAMP)));
79+
}
4880
}
Lines changed: 107 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,107 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
*
4+
* The OpenSearch Contributors require contributions made to
5+
* this file be licensed under the Apache-2.0 license or a
6+
* compatible open source license.
7+
*/
8+
9+
package org.opensearch.analytics.qa;
10+
11+
import org.opensearch.client.Request;
12+
import org.opensearch.client.Response;
13+
14+
import java.util.List;
15+
import java.util.Map;
16+
17+
/** End-to-end test for late-materialization queries against a {@code date_nanos} timestamp on a multi-shard parquet index. */
18+
public class LateMaterializationDateNanosIT extends AnalyticsRestTestCase {
19+
20+
private static final String INDEX = "late_mat_date_nanos_e2e";
21+
private static final int NUM_SHARDS = 2;
22+
23+
public void testMultiShardLateMatSortFetchOnDateNanos() throws Exception {
24+
createParquetBackedIndex();
25+
indexDocs();
26+
27+
// `severity` is fetch-only (projected, not in filter/sort) — required for the LM rewriter
28+
// to fire; otherwise above ⊆ below and it skips, masking the bug.
29+
Map<String, Object> result = executePpl(
30+
"source = " + INDEX
31+
+ " | where match(body, 'failed') and service = 'checkout'"
32+
+ " | sort - ts"
33+
+ " | fields ts, severity, body"
34+
+ " | head 4"
35+
);
36+
37+
List<String> columns = extractColumnNames(result);
38+
assertTrue("schema must contain 'ts', got " + columns, columns.contains("ts"));
39+
assertTrue("schema must contain 'severity', got " + columns, columns.contains("severity"));
40+
assertTrue("schema must contain 'body', got " + columns, columns.contains("body"));
41+
42+
@SuppressWarnings("unchecked")
43+
List<List<Object>> rows = (List<List<Object>>) result.get("datarows");
44+
assertNotNull("Stitcher copyFromSafe trip returns no payload", rows);
45+
assertEquals("4 matching docs across 2 shards", 4, rows.size());
46+
47+
int tsIdx = columns.indexOf("ts");
48+
String firstTs = String.valueOf(rows.get(0).get(tsIdx));
49+
String lastTs = String.valueOf(rows.get(rows.size() - 1).get(tsIdx));
50+
assertTrue("DESC: first " + firstTs + " >= last " + lastTs, firstTs.compareTo(lastTs) >= 0);
51+
52+
// sub-ms precision survives — pre-fix Stitcher silently coerces to ms (3 digits).
53+
assertTrue("expected >3 fractional digits in " + firstTs, firstTs.matches(".*\\.\\d{4,}.*"));
54+
}
55+
56+
private void createParquetBackedIndex() throws Exception {
57+
try {
58+
client().performRequest(new Request("DELETE", "/" + INDEX));
59+
} catch (Exception ignored) {}
60+
61+
String body = "{"
62+
+ "\"settings\": {"
63+
+ " \"number_of_shards\": " + NUM_SHARDS + ","
64+
+ " \"number_of_replicas\": 0,"
65+
+ " \"index.pluggable.dataformat.enabled\": true,"
66+
+ " \"index.pluggable.dataformat\": \"composite\","
67+
+ " \"index.composite.primary_data_format\": \"parquet\","
68+
+ " \"index.composite.secondary_data_formats\": \"lucene\""
69+
+ "},"
70+
+ "\"mappings\": {"
71+
+ " \"properties\": {"
72+
+ " \"ts\": { \"type\": \"date_nanos\" },"
73+
+ " \"service\": { \"type\": \"keyword\" },"
74+
+ " \"severity\": { \"type\": \"keyword\" },"
75+
+ " \"body\": { \"type\": \"text\", \"store\": true }"
76+
+ " }"
77+
+ "}"
78+
+ "}";
79+
80+
Request createIndex = new Request("PUT", "/" + INDEX);
81+
createIndex.setJsonEntity(body);
82+
Response response = client().performRequest(createIndex);
83+
assertOkAndParse(response, "Create index " + INDEX);
84+
}
85+
86+
private void indexDocs() throws Exception {
87+
String bulk =
88+
"{\"index\":{\"_index\":\"" + INDEX + "\",\"_id\":\"1\"}}\n"
89+
+ "{\"ts\":\"2025-09-23T00:01:01.123456Z\",\"service\":\"checkout\",\"severity\":\"ERROR\",\"body\":\"failed order: payment declined\"}\n"
90+
+ "{\"index\":{\"_index\":\"" + INDEX + "\",\"_id\":\"2\"}}\n"
91+
+ "{\"ts\":\"2025-09-23T00:02:01.234567Z\",\"service\":\"checkout\",\"severity\":\"ERROR\",\"body\":\"failed order due to expired session\"}\n"
92+
+ "{\"index\":{\"_index\":\"" + INDEX + "\",\"_id\":\"3\"}}\n"
93+
+ "{\"ts\":\"2025-09-23T00:03:01.345678Z\",\"service\":\"checkout\",\"severity\":\"WARN\",\"body\":\"failed order: inventory check\"}\n"
94+
+ "{\"index\":{\"_index\":\"" + INDEX + "\",\"_id\":\"4\"}}\n"
95+
+ "{\"ts\":\"2025-09-23T00:04:01.456789Z\",\"service\":\"checkout\",\"severity\":\"ERROR\",\"body\":\"failed gateway\"}\n"
96+
+ "{\"index\":{\"_index\":\"" + INDEX + "\",\"_id\":\"5\"}}\n"
97+
+ "{\"ts\":\"2025-09-23T00:05:00.000000Z\",\"service\":\"frontend\",\"severity\":\"INFO\",\"body\":\"ok\"}\n"
98+
+ "{\"index\":{\"_index\":\"" + INDEX + "\",\"_id\":\"6\"}}\n"
99+
+ "{\"ts\":\"2025-09-23T00:06:00.000000Z\",\"service\":\"checkout\",\"severity\":\"INFO\",\"body\":\"successful\"}\n";
100+
Request bulkReq = new Request("POST", "/_bulk");
101+
bulkReq.addParameter("refresh", "true");
102+
bulkReq.setJsonEntity(bulk);
103+
assertOkAndParse(client().performRequest(bulkReq), "Bulk index " + INDEX);
104+
105+
client().performRequest(new Request("POST", "/" + INDEX + "/_flush?force=true"));
106+
}
107+
}

0 commit comments

Comments
 (0)