> {
+ if observed.len() != template.len() {
+ return plan_err!(
+ "{udf_name} expects {} arguments, got {}",
+ template.len(),
+ observed.len()
+ );
+ }
+ template
+ .iter()
+ .enumerate()
+ .map(|(i, mode)| coerce_slot(udf_name, i, &observed[i], *mode))
+ .collect()
+}
+
+pub mod convert_tz;
+
+pub fn register_all(ctx: &SessionContext) {
+ convert_tz::register_all(ctx);
+ log::info!("OpenSearch UDF register_all: convert_tz registered");
+}
+
+#[cfg(test)]
+mod tests {
+ //! Direct tests for the [`CoerceMode`] helper library. `convert_tz` exercises
+ //! `TimestampMs` and `Utf8` through its public `coerce_types`; these tests
+ //! cover every mode's accept + reject paths so future UDFs that pick up
+ //! `Date32`, `Int64`, or `Float64` inherit a proven helper rather than being
+ //! the first caller.
+ use super::{coerce_args, coerce_slot, CoerceMode};
+ use datafusion::arrow::datatypes::{DataType, TimeUnit};
+
+ fn ts_ms() -> DataType {
+ DataType::Timestamp(TimeUnit::Millisecond, None)
+ }
+
+ // ── TimestampMs ────────────────────────────────────────────────────────
+ #[test]
+ fn timestampms_accepts_every_temporal_source() {
+ for observed in [
+ DataType::Utf8,
+ DataType::LargeUtf8,
+ DataType::Utf8View,
+ DataType::Date32,
+ DataType::Date64,
+ DataType::Timestamp(TimeUnit::Second, None),
+ DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
+ ] {
+ let result = coerce_slot("t", 0, &observed, CoerceMode::TimestampMs).unwrap();
+ assert_eq!(result, ts_ms(), "TimestampMs should canonicalize {observed:?}");
+ }
+ }
+
+ #[test]
+ fn timestampms_rejects_numeric() {
+ let err = coerce_slot("t", 0, &DataType::Int64, CoerceMode::TimestampMs).unwrap_err();
+ assert!(err.to_string().contains("expected timestamp/date/string"));
+ }
+
+ // ── Date32 ─────────────────────────────────────────────────────────────
+ #[test]
+ fn date32_accepts_date_and_string_sources() {
+ for observed in [
+ DataType::Date32,
+ DataType::Date64,
+ DataType::Utf8,
+ DataType::LargeUtf8,
+ DataType::Utf8View,
+ DataType::Timestamp(TimeUnit::Millisecond, None),
+ ] {
+ let result = coerce_slot("d", 0, &observed, CoerceMode::Date32).unwrap();
+ assert_eq!(result, DataType::Date32);
+ }
+ }
+
+ #[test]
+ fn date32_rejects_numeric() {
+ let err = coerce_slot("d", 0, &DataType::Float64, CoerceMode::Date32).unwrap_err();
+ assert!(err.to_string().contains("expected date/timestamp/string"));
+ }
+
+ // ── Int64 ──────────────────────────────────────────────────────────────
+ #[test]
+ fn int64_accepts_every_number() {
+ for observed in [
+ DataType::Int8,
+ DataType::Int16,
+ DataType::Int32,
+ DataType::Int64,
+ DataType::UInt8,
+ DataType::UInt16,
+ DataType::UInt32,
+ DataType::UInt64,
+ DataType::Float32,
+ DataType::Float64,
+ ] {
+ let result = coerce_slot("i", 0, &observed, CoerceMode::Int64).unwrap();
+ assert_eq!(result, DataType::Int64);
+ }
+ }
+
+ #[test]
+ fn int64_rejects_strings() {
+ let err = coerce_slot("i", 0, &DataType::Utf8, CoerceMode::Int64).unwrap_err();
+ assert!(err.to_string().contains("expected integer or float"));
+ }
+
+ // ── Float64 ────────────────────────────────────────────────────────────
+ #[test]
+ fn float64_accepts_every_number() {
+ for observed in [
+ DataType::Int32,
+ DataType::Int64,
+ DataType::UInt32,
+ DataType::Float32,
+ DataType::Float64,
+ ] {
+ let result = coerce_slot("f", 0, &observed, CoerceMode::Float64).unwrap();
+ assert_eq!(result, DataType::Float64);
+ }
+ }
+
+ #[test]
+ fn float64_rejects_strings() {
+ let err = coerce_slot("f", 0, &DataType::Utf8, CoerceMode::Float64).unwrap_err();
+ assert!(err.to_string().contains("expected integer or float"));
+ }
+
+ // ── Utf8 ───────────────────────────────────────────────────────────────
+ #[test]
+ fn utf8_accepts_every_string_variant() {
+ for observed in [DataType::Utf8, DataType::LargeUtf8, DataType::Utf8View] {
+ let result = coerce_slot("s", 0, &observed, CoerceMode::Utf8).unwrap();
+ assert_eq!(result, DataType::Utf8);
+ }
+ }
+
+ #[test]
+ fn utf8_rejects_numeric_and_temporal() {
+ for observed in [DataType::Int64, DataType::Float64, DataType::Date32] {
+ let err = coerce_slot("s", 0, &observed, CoerceMode::Utf8).unwrap_err();
+ assert!(err.to_string().contains("expected string"));
+ }
+ }
+
+ // ── coerce_args ────────────────────────────────────────────────────────
+ #[test]
+ fn coerce_args_maps_each_slot_through_its_mode() {
+ let observed = [DataType::Utf8, DataType::Int32];
+ let template = [CoerceMode::TimestampMs, CoerceMode::Int64];
+ let result = coerce_args("multi", &observed, &template).unwrap();
+ assert_eq!(result, vec![ts_ms(), DataType::Int64]);
+ }
+
+ #[test]
+ fn coerce_args_rejects_arity_mismatch() {
+ let observed = [DataType::Utf8];
+ let template = [CoerceMode::Utf8, CoerceMode::Utf8];
+ let err = coerce_args("arity", &observed, &template).unwrap_err();
+ assert!(err.to_string().contains("expects 2 arguments, got 1"));
+ }
+
+ #[test]
+ fn coerce_args_propagates_slot_errors() {
+ let observed = [DataType::Utf8, DataType::Utf8];
+ let template = [CoerceMode::Utf8, CoerceMode::Int64];
+ let err = coerce_args("slot", &observed, &template).unwrap_err();
+ assert!(
+ err.to_string().contains("arg 1"),
+ "error must name the failing slot index, got: {err}"
+ );
+ }
+}
diff --git a/sandbox/plugins/analytics-backend-datafusion/src/internalClusterTest/java/org/opensearch/be/datafusion/BaseScalarFunctionIT.java b/sandbox/plugins/analytics-backend-datafusion/src/internalClusterTest/java/org/opensearch/be/datafusion/BaseScalarFunctionIT.java
new file mode 100644
index 0000000000000..6dd4a9116aaee
--- /dev/null
+++ b/sandbox/plugins/analytics-backend-datafusion/src/internalClusterTest/java/org/opensearch/be/datafusion/BaseScalarFunctionIT.java
@@ -0,0 +1,204 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.be.datafusion;
+
+import org.opensearch.Version;
+import org.opensearch.action.admin.indices.create.CreateIndexResponse;
+import org.opensearch.analytics.AnalyticsPlugin;
+import org.opensearch.arrow.flight.transport.FlightStreamPlugin;
+import org.opensearch.be.lucene.LucenePlugin;
+import org.opensearch.cluster.metadata.IndexMetadata;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.common.util.FeatureFlags;
+import org.opensearch.common.xcontent.XContentFactory;
+import org.opensearch.composite.CompositeDataFormatPlugin;
+import org.opensearch.core.xcontent.XContentBuilder;
+import org.opensearch.parquet.ParquetDataFormatPlugin;
+import org.opensearch.plugins.Plugin;
+import org.opensearch.plugins.PluginInfo;
+import org.opensearch.ppl.TestPPLPlugin;
+import org.opensearch.ppl.action.PPLRequest;
+import org.opensearch.ppl.action.PPLResponse;
+import org.opensearch.ppl.action.UnifiedPPLExecuteAction;
+import org.opensearch.test.OpenSearchIntegTestCase;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Shared fixture + scalar-result assert helpers for end-to-end PPL → Calcite →
+ * Substrait → DataFusion scalar-function tests.
+ *
+ * Each subclass declares its functions as test methods using the
+ * {@code assertScalarXxx(expr, expected)} helpers. The query template is fixed:
+ * {@code source=bank | eval x = | fields x | head 1}. Inputs are
+ * literals so assertions don't depend on the bank fixture's data — the test
+ * exercises the function's name lookup, type inference, and runtime, not
+ * arithmetic on rows.
+ *
+ * @opensearch.internal
+ */
+// TEST-scope cluster per method — slower but eliminates cluster-reuse degradation that
+// surfaces as cascading NodeDisconnectedException when many test methods share a SUITE cluster.
+@OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.SUITE, numDataNodes = 1)
+public abstract class BaseScalarFunctionIT extends OpenSearchIntegTestCase {
+
+ protected static final String BANK_INDEX = "bank";
+
+ @Override
+ protected Collection> nodePlugins() {
+ return List.of(TestPPLPlugin.class, FlightStreamPlugin.class, CompositeDataFormatPlugin.class, LucenePlugin.class);
+ }
+
+ @Override
+ protected Collection additionalNodePlugins() {
+ return List.of(
+ classpathPlugin(AnalyticsPlugin.class, Collections.emptyList()),
+ classpathPlugin(ParquetDataFormatPlugin.class, Collections.emptyList()),
+ classpathPlugin(DataFusionPlugin.class, List.of(AnalyticsPlugin.class.getName()))
+ );
+ }
+
+ private static PluginInfo classpathPlugin(Class pluginClass, List extendedPlugins) {
+ return new PluginInfo(
+ pluginClass.getName(),
+ "classpath plugin",
+ "NA",
+ Version.CURRENT,
+ "1.8",
+ pluginClass.getName(),
+ null,
+ extendedPlugins,
+ false
+ );
+ }
+
+ @Override
+ protected Settings nodeSettings(int nodeOrdinal) {
+ return Settings.builder()
+ .put(super.nodeSettings(nodeOrdinal))
+ .put(FeatureFlags.PLUGGABLE_DATAFORMAT_EXPERIMENTAL_FLAG, true)
+ .build();
+ }
+
+ @Override
+ public void setUp() throws Exception {
+ super.setUp();
+ // SUITE-scoped cluster is reused across test methods — only create/index once.
+ if (!indexExists(BANK_INDEX)) {
+ createBankIndex();
+ indexBankDocs();
+ ensureGreen(BANK_INDEX);
+ refresh(BANK_INDEX);
+ }
+ }
+
+ private void createBankIndex() throws Exception {
+ XContentBuilder mapping = XContentFactory.jsonBuilder()
+ .startObject()
+ .startObject("properties")
+ .startObject("account_number")
+ .field("type", "long")
+ .endObject()
+ .startObject("firstname")
+ .field("type", "keyword")
+ .endObject()
+ .startObject("balance")
+ .field("type", "long")
+ .endObject()
+ .startObject("created_at")
+ .field("type", "date")
+ .endObject()
+ .endObject()
+ .endObject();
+
+ Settings indexSettings = Settings.builder()
+ .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1)
+ .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
+ .put("index.pluggable.dataformat.enabled", true)
+ .put("index.pluggable.dataformat", "composite")
+ .put("index.composite.primary_data_format", "parquet")
+ .putList("index.composite.secondary_data_formats")
+ .build();
+
+ CreateIndexResponse response = client().admin()
+ .indices()
+ .prepareCreate(BANK_INDEX)
+ .setSettings(indexSettings)
+ .setMapping(mapping)
+ .get();
+ assertTrue("bank index creation must be acknowledged", response.isAcknowledged());
+ }
+
+ private void indexBankDocs() {
+ client().prepareIndex(BANK_INDEX)
+ .setId("1")
+ .setSource("account_number", 1, "firstname", "Amber", "balance", 39225L, "created_at", "2024-06-15T10:30:00Z")
+ .get();
+ client().prepareIndex(BANK_INDEX)
+ .setId("6")
+ .setSource("account_number", 6, "firstname", "Hattie", "balance", 5686L, "created_at", "2024-01-20T14:45:30Z")
+ .get();
+ }
+
+ // ---- Assert helpers ----
+
+ /**
+ * Runs the given expression against the single bank row with
+ * {@code account_number=1} (firstname='Amber', balance=39225) and returns
+ * the resulting cell. Pinning the row makes assertions deterministic and
+ * lets tests reference {@code firstname} / {@code balance} as fields —
+ * which prevents Calcite's constant-folding from optimizing the function
+ * away at plan time. Tests must therefore use field references to truly
+ * exercise the Substrait + DataFusion runtime path.
+ */
+ protected Object evalScalar(String expr) {
+ PPLRequest request = new PPLRequest(
+ "source=" + BANK_INDEX + " | where account_number = 1 | eval x = " + expr + " | fields x | head 1"
+ );
+ PPLResponse response = client().execute(UnifiedPPLExecuteAction.INSTANCE, request).actionGet();
+ assertNotNull("PPLResponse must not be null", response);
+ assertEquals("schema columns", List.of("x"), response.getColumns());
+ assertEquals("head 1 → exactly 1 row", 1, response.getRows().size());
+ return response.getRows().get(0)[0];
+ }
+
+ protected void assertScalarLong(String expr, long expected) {
+ Object cell = evalScalar(expr);
+ assertNotNull(expr + " result must not be null", cell);
+ assertTrue(expr + " result must be Number, got " + cell.getClass(), cell instanceof Number);
+ assertEquals(expr, expected, ((Number) cell).longValue());
+ }
+
+ protected void assertScalarDouble(String expr, double expected, double delta) {
+ Object cell = evalScalar(expr);
+ assertNotNull(expr + " result must not be null", cell);
+ assertTrue(expr + " result must be Number, got " + cell.getClass(), cell instanceof Number);
+ assertEquals(expr, expected, ((Number) cell).doubleValue(), delta);
+ }
+
+ protected void assertScalarString(String expr, String expected) {
+ Object cell = evalScalar(expr);
+ assertNotNull(expr + " result must not be null", cell);
+ assertEquals(expr, expected, cell.toString());
+ }
+
+ protected void assertScalarBoolean(String expr, boolean expected) {
+ Object cell = evalScalar(expr);
+ assertNotNull(expr + " result must not be null", cell);
+ assertTrue(expr + " result must be Boolean, got " + cell.getClass(), cell instanceof Boolean);
+ assertEquals(expr, expected, cell);
+ }
+
+ protected void assertScalarNull(String expr) {
+ Object cell = evalScalar(expr);
+ assertNull(expr + " result must be null but was " + cell, cell);
+ }
+}
diff --git a/sandbox/plugins/analytics-backend-datafusion/src/internalClusterTest/java/org/opensearch/be/datafusion/ScalarDateTimeFunctionIT.java b/sandbox/plugins/analytics-backend-datafusion/src/internalClusterTest/java/org/opensearch/be/datafusion/ScalarDateTimeFunctionIT.java
new file mode 100644
index 0000000000000..4729a89663b58
--- /dev/null
+++ b/sandbox/plugins/analytics-backend-datafusion/src/internalClusterTest/java/org/opensearch/be/datafusion/ScalarDateTimeFunctionIT.java
@@ -0,0 +1,42 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.be.datafusion;
+
+/**
+ * End-to-end smoke tests for scalar date/time functions routed through PPL → Calcite →
+ * Substrait → DataFusion. Bank fixture row 1: created_at='2024-06-15T10:30:00Z'.
+ *
+ * Two representative cases:
+ *
+ * - {@link #testYear()} — YAML alias with literal-arg injection
+ * ({@code YEAR(ts) → date_part('year', ts)}).
+ * - {@link #testConvertTz()} — custom Rust UDF registered with DataFusion
+ * ({@code convert_tz(ts, from_tz, to_tz)}).
+ *
+ */
+public class ScalarDateTimeFunctionIT extends BaseScalarFunctionIT {
+
+ public void testYear() {
+ Object cell = evalScalar("year(created_at)");
+ assertNotNull("year() must not be null", cell);
+ assertEquals(2024L, ((Number) cell).longValue());
+ }
+
+ public void testConvertTz() {
+ // row 1: created_at = 2024-06-15T10:30:00Z (UTC).
+ // Shifted UTC → +10:00 = 2024-06-15T20:30:00Z, unix seconds = 1718483400.
+ // ConvertTzAdapter rewrites PPL's bespoke CONVERT_TZ to our locally-declared
+ // SqlFunction("convert_tz") whose Sig is in ADDITIONAL_SCALAR_SIGS;
+ // UnixTimestampAdapter does the same to to_unixtime. Isthmus resolves both,
+ // DataFusion runs convert_tz via the Rust UDF and to_unixtime natively.
+ Object cell = evalScalar("unix_timestamp(convert_tz(created_at, '+00:00', '+10:00'))");
+ assertNotNull("convert_tz must not be null", cell);
+ assertEquals(1718483400L, ((Number) cell).longValue());
+ }
+}
diff --git a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/ConvertTzAdapter.java b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/ConvertTzAdapter.java
new file mode 100644
index 0000000000000..cb460333dbdaa
--- /dev/null
+++ b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/ConvertTzAdapter.java
@@ -0,0 +1,191 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.be.datafusion;
+
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.SqlFunction;
+import org.apache.calcite.sql.SqlFunctionCategory;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.type.OperandTypes;
+import org.apache.calcite.sql.type.ReturnTypes;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.opensearch.analytics.spi.AbstractNameMappingAdapter;
+import org.opensearch.analytics.spi.FieldStorageInfo;
+import org.opensearch.analytics.spi.ScalarFunctionAdapter;
+
+import java.time.DateTimeException;
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Locale;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * Cat-3b adapter for PPL's {@code CONVERT_TZ(ts, from_tz, to_tz)}. Two jobs in
+ * priority order:
+ *
+ *
+ * - Identity short-circuit: when both tz operands are string
+ * literals and canonicalize to the same value, the call reduces to its
+ * timestamp operand. No UDF invocation, no wire traffic.
+ * - UDF fallback with canonicalized literal operands: every other
+ * case rewrites to {@link #LOCAL_CONVERT_TZ_OP} whose
+ * {@code FunctionMappings.Sig} in {@link DataFusionFragmentConvertor}
+ * resolves to the {@code convert_tz} Rust UDF. Literal tz operands are
+ * validated + canonicalized via {@link #canonicalizeTz(String)} at plan
+ * time so bad literals surface with a clear error rather than silent
+ * per-row NULL at runtime.
+ *
+ *
+ * Why no offset+offset → interval fold: building an interval literal at
+ * Calcite's level requires {@code org.apache.calcite.avatica.util.TimeUnit},
+ * which lives in avatica and is a {@code runtimeOnly} dep of this module.
+ * Pulling it in just for the fixed-offset case doesn't pay for itself; IANA
+ * pairs dominate real-world {@code CONVERT_TZ} usage and must go through the
+ * UDF anyway (per-row DST lookup).
+ *
+ *
The fallback preserves the original call's return type via
+ * {@code rexBuilder.makeCall(original.getType(), ...)} so the enclosing
+ * {@code Project} / {@code Filter} rowType cache stays consistent (see
+ * {@link AbstractNameMappingAdapter} javadoc for background).
+ *
+ * @opensearch.internal
+ */
+class ConvertTzAdapter implements ScalarFunctionAdapter {
+
+ /**
+ * Locally-declared target operator for the rewrite. {@link SqlKind#OTHER_FUNCTION}
+ * so it doesn't collide with any Calcite built-in.
+ * {@link OperandTypes#ANY_STRING_STRING} keeps validation permissive on the
+ * timestamp slot — real argument vetting happens inside the UDF's
+ * {@code coerce_types} and {@code invoke_with_args}.
+ */
+ static final SqlOperator LOCAL_CONVERT_TZ_OP = new SqlFunction(
+ "convert_tz",
+ SqlKind.OTHER_FUNCTION,
+ ReturnTypes.ARG0_NULLABLE,
+ null,
+ OperandTypes.ANY_STRING_STRING,
+ SqlFunctionCategory.TIMEDATE
+ );
+
+ /** Matches {@code ±H:MM} / {@code ±HH:MM} with hours [0,14] and minutes [0,59]. */
+ private static final Pattern OFFSET_PATTERN = Pattern.compile("^([+-])(\\d{1,2}):(\\d{2})$");
+
+ @Override
+ public RexNode adapt(RexCall original, List fieldStorage, RelOptCluster cluster) {
+ RexBuilder rexBuilder = cluster.getRexBuilder();
+ List operands = new ArrayList<>(original.getOperands());
+ // Slot 0 is the timestamp; slots 1 and 2 are from_tz / to_tz.
+ for (int slot : new int[] { 1, 2 }) {
+ operands.set(slot, canonicalizeTzOperand(operands.get(slot), rexBuilder));
+ }
+
+ // Identity short-circuit: both operands resolve to the same canonical
+ // string → the conversion is a no-op.
+ String fromLiteral = tzLiteralValue(operands.get(1));
+ String toLiteral = tzLiteralValue(operands.get(2));
+ if (fromLiteral != null && toLiteral != null && fromLiteral.equals(toLiteral)) {
+ return operands.get(0);
+ }
+
+ // UDF fallback. Preserve the original call's return type — see
+ // AbstractNameMappingAdapter for why (Project.isValid compatibleTypes check).
+ return rexBuilder.makeCall(original.getType(), LOCAL_CONVERT_TZ_OP, operands);
+ }
+
+ /**
+ * Returns the string value of a canonicalized tz literal operand, or null
+ * when the operand is not a VARCHAR/CHAR {@link RexLiteral} (column refs,
+ * NULL literals, other expressions).
+ */
+ private static String tzLiteralValue(RexNode operand) {
+ if (!(operand instanceof RexLiteral literal)) return null;
+ SqlTypeName typeName = literal.getType().getSqlTypeName();
+ if (typeName != SqlTypeName.CHAR && typeName != SqlTypeName.VARCHAR) return null;
+ return literal.getValueAs(String.class);
+ }
+
+ /**
+ * If {@code operand} is a string {@link RexLiteral}, canonicalize it and
+ * return a new literal with the canonical form (or the original if already
+ * canonical). Non-literal operands (column references, function results)
+ * pass through untouched — their runtime values can't be validated until
+ * the UDF runs.
+ *
+ * Throws {@link IllegalArgumentException} for literals that don't match
+ * either the {@code ±HH:MM} offset pattern or a known IANA zone id.
+ */
+ private static RexNode canonicalizeTzOperand(RexNode operand, RexBuilder rexBuilder) {
+ if (!(operand instanceof RexLiteral literal)) {
+ return operand;
+ }
+ SqlTypeName typeName = literal.getType().getSqlTypeName();
+ if (typeName != SqlTypeName.CHAR && typeName != SqlTypeName.VARCHAR) {
+ return operand;
+ }
+ String raw = literal.getValueAs(String.class);
+ if (raw == null) {
+ // NULL literal — UDF handles null operand at runtime.
+ return operand;
+ }
+ String canonical = canonicalizeTz(raw);
+ if (canonical.equals(raw)) {
+ return operand;
+ }
+ return rexBuilder.makeLiteral(
+ canonical,
+ rexBuilder.getTypeFactory().createSqlType(SqlTypeName.VARCHAR),
+ literal.getType().isNullable()
+ );
+ }
+
+ /**
+ * Canonicalize a timezone string. Accepts either:
+ *
+ * - {@code ±H:MM} / {@code ±HH:MM} where hours ∈ [0,14] and minutes ∈ [0,59];
+ * returned zero-padded as {@code ±HH:MM}.
+ * - IANA zone id recognized by {@link ZoneId#of(String)}; returned as the
+ * JDK-normalized form. {@code ZoneId.of} rejects unknown ids, so invalid
+ * IANA names surface here as {@link IllegalArgumentException}.
+ *
+ *
+ * The {@code ±HH:MM} bounds match the Rust UDF's {@code parse_offset_seconds}
+ * (rust/src/udf/convert_tz.rs) — `+14:59` is the maximum offset anywhere on
+ * Earth (Kiribati is +14:00; the extra minute tolerance matches existing
+ * UDF behavior).
+ */
+ static String canonicalizeTz(String raw) {
+ Matcher offset = OFFSET_PATTERN.matcher(raw);
+ if (offset.matches()) {
+ String sign = offset.group(1);
+ int hours = Integer.parseInt(offset.group(2));
+ int minutes = Integer.parseInt(offset.group(3));
+ if (hours > 14 || minutes > 59) {
+ throw new IllegalArgumentException(
+ "convert_tz: invalid offset [" + raw + "] — hours must be in [0, 14] and minutes in [0, 59]"
+ );
+ }
+ return String.format(Locale.ROOT, "%s%02d:%02d", sign, hours, minutes);
+ }
+ try {
+ // ZoneId.of() throws for unknown ids; the returned ZoneId.getId()
+ // is the JDK's canonical form (same id for equivalent inputs).
+ return ZoneId.of(raw).getId();
+ } catch (DateTimeException e) {
+ throw new IllegalArgumentException("convert_tz: invalid timezone [" + raw + "] — expected IANA zone id or ±HH:MM offset", e);
+ }
+ }
+}
diff --git a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java
index 221838a98dece..1119f8d8fa17a 100644
--- a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java
+++ b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java
@@ -100,7 +100,10 @@ public class DataFusionAnalyticsBackendPlugin implements AnalyticsSearchBackendP
ScalarFunction.MINUS,
ScalarFunction.TIMES,
ScalarFunction.DIVIDE,
- ScalarFunction.MOD
+ ScalarFunction.MOD,
+ ScalarFunction.YEAR,
+ ScalarFunction.CONVERT_TZ,
+ ScalarFunction.UNIX_TIMESTAMP
);
private static final Set AGG_FUNCTIONS = Set.of(
@@ -178,7 +181,10 @@ public Map scalarFunctionAdapters() {
Map.entry(ScalarFunction.SARG_PREDICATE, new SargAdapter()),
Map.entry(ScalarFunction.DIVIDE, new StdOperatorRewriteAdapter("DIVIDE", SqlStdOperatorTable.DIVIDE)),
Map.entry(ScalarFunction.MOD, new StdOperatorRewriteAdapter("MOD", SqlStdOperatorTable.MOD)),
- Map.entry(ScalarFunction.LIKE, new LikeAdapter())
+ Map.entry(ScalarFunction.LIKE, new LikeAdapter()),
+ Map.entry(ScalarFunction.YEAR, new YearAdapter()),
+ Map.entry(ScalarFunction.CONVERT_TZ, new ConvertTzAdapter()),
+ Map.entry(ScalarFunction.UNIX_TIMESTAMP, new UnixTimestampAdapter())
);
}
};
diff --git a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionFragmentConvertor.java b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionFragmentConvertor.java
index fe072c85a2e99..e7d67bb5879cf 100644
--- a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionFragmentConvertor.java
+++ b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionFragmentConvertor.java
@@ -89,7 +89,11 @@ public class DataFusionFragmentConvertor implements FragmentConvertor {
*/
private static final List ADDITIONAL_SCALAR_SIGS = List.of(
FunctionMappings.s(DelegatedPredicateFunction.FUNCTION, DelegatedPredicateFunction.NAME),
- FunctionMappings.s(SqlLibraryOperators.ILIKE, "ilike")
+ FunctionMappings.s(SqlLibraryOperators.ILIKE, "ilike"),
+ FunctionMappings.s(DelegatedPredicateFunction.FUNCTION, DelegatedPredicateFunction.NAME),
+ FunctionMappings.s(SqlLibraryOperators.DATE_PART, "date_part"),
+ FunctionMappings.s(ConvertTzAdapter.LOCAL_CONVERT_TZ_OP, "convert_tz"),
+ FunctionMappings.s(UnixTimestampAdapter.LOCAL_TO_UNIXTIME_OP, "to_unixtime")
);
private final SimpleExtension.ExtensionCollection extensions;
diff --git a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/UnixTimestampAdapter.java b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/UnixTimestampAdapter.java
new file mode 100644
index 0000000000000..2f7056ac92c55
--- /dev/null
+++ b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/UnixTimestampAdapter.java
@@ -0,0 +1,60 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.be.datafusion;
+
+import org.apache.calcite.sql.SqlFunction;
+import org.apache.calcite.sql.SqlFunctionCategory;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.type.OperandTypes;
+import org.apache.calcite.sql.type.ReturnTypes;
+import org.opensearch.analytics.spi.AbstractNameMappingAdapter;
+
+import java.util.List;
+
+/**
+ * Cat-3a rename adapter for PPL's {@code UNIX_TIMESTAMP(ts)}. Rewrites to a
+ * locally-declared {@link SqlFunction} named {@code to_unixtime} — the name
+ * DataFusion's substrait consumer recognizes for its native
+ * {@code ToUnixtimeFunc} (no UDF registration required on the Rust side).
+ *
+ * Same machinery as {@link ConvertTzAdapter}: locally-declared operator is
+ * the referent of the {@link io.substrait.isthmus.expression.FunctionMappings.Sig}
+ * in {@link DataFusionFragmentConvertor#ADDITIONAL_SCALAR_SIGS}.
+ *
+ *
Type note. PPL's {@code UNIX_TIMESTAMP} returns
+ * {@code DOUBLE_FORCE_NULLABLE}; DataFusion's {@code to_unixtime} returns
+ * {@code Int64}. {@link AbstractNameMappingAdapter} preserves the PPL-declared
+ * return type on the rewritten call so Calcite's {@code Project.isValid}
+ * assertion holds. The downstream substrait consumer (DataFusion) re-resolves
+ * {@code to_unixtime} by name and applies its own {@code coerce_types}, so the
+ * Calcite-inferred type is purely plan-validity bookkeeping.
+ *
+ * @opensearch.internal
+ */
+class UnixTimestampAdapter extends AbstractNameMappingAdapter {
+
+ /**
+ * Locally-declared target operator. Name matches DataFusion's native
+ * {@code to_unixtime}. Return-type inference is irrelevant — the adapter
+ * clones with the original PPL return type.
+ */
+ static final SqlOperator LOCAL_TO_UNIXTIME_OP = new SqlFunction(
+ "to_unixtime",
+ SqlKind.OTHER_FUNCTION,
+ ReturnTypes.BIGINT_NULLABLE,
+ null,
+ OperandTypes.ANY,
+ SqlFunctionCategory.TIMEDATE
+ );
+
+ UnixTimestampAdapter() {
+ super(LOCAL_TO_UNIXTIME_OP, List.of(), List.of());
+ }
+}
diff --git a/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/YearAdapter.java b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/YearAdapter.java
new file mode 100644
index 0000000000000..5ad28fc0ba13a
--- /dev/null
+++ b/sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/YearAdapter.java
@@ -0,0 +1,33 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.be.datafusion;
+
+import org.apache.calcite.sql.fun.SqlLibraryOperators;
+import org.opensearch.analytics.spi.AbstractNameMappingAdapter;
+
+import java.util.List;
+
+/**
+ * Representative {@link AbstractNameMappingAdapter} for Calcite {@code YEAR(ts)}.
+ * Rewrites to {@code date_part('year', ts)} so isthmus resolves it against
+ * DataFusion's native date_part (see the {@code date_part} signature in
+ * {@code opensearch_scalar.yaml}). Demonstrates the reusable rename +
+ * literal-arg-injection adapter pattern for cat-3 PPL functions.
+ *
+ *
Follow-up PRs extend the pattern to MONTH/DAY/HOUR/etc. each as a
+ * one-line concrete subclass — identical shape, different unit literal.
+ *
+ * @opensearch.internal
+ */
+class YearAdapter extends AbstractNameMappingAdapter {
+
+ YearAdapter() {
+ super(SqlLibraryOperators.DATE_PART, List.of("year"), List.of());
+ }
+}
diff --git a/sandbox/plugins/analytics-backend-datafusion/src/main/resources/opensearch_scalar_functions.yaml b/sandbox/plugins/analytics-backend-datafusion/src/main/resources/opensearch_scalar_functions.yaml
index 4889f35c11b6a..8f5d778e543b3 100644
--- a/sandbox/plugins/analytics-backend-datafusion/src/main/resources/opensearch_scalar_functions.yaml
+++ b/sandbox/plugins/analytics-backend-datafusion/src/main/resources/opensearch_scalar_functions.yaml
@@ -20,3 +20,21 @@ scalar_functions:
- value: "string"
name: "match"
return: boolean
+ - name: "date_part"
+ impls:
+ - args: [{ value: string, name: "part" }, { value: "any1", name: "value" }]
+ return: any1
+
+ - name: "convert_tz"
+ description: "Shift a timestamp from one timezone to another. IANA names and +/-HH:MM offsets."
+ impls:
+ - args:
+ - { value: "any1", name: "ts" }
+ - { value: string, name: "from_tz" }
+ - { value: string, name: "to_tz" }
+ return: any1
+ - name: "to_unixtime"
+ description: "Return a timestamp as Unix epoch seconds."
+ impls:
+ - args: [{ value: "any1", name: "ts" }]
+ return: any1
diff --git a/sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/ConvertTzAdapterTests.java b/sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/ConvertTzAdapterTests.java
new file mode 100644
index 0000000000000..19eb0df9ad578
--- /dev/null
+++ b/sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/ConvertTzAdapterTests.java
@@ -0,0 +1,228 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.be.datafusion;
+
+import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.hep.HepPlanner;
+import org.apache.calcite.plan.hep.HepProgramBuilder;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.SqlFunction;
+import org.apache.calcite.sql.SqlFunctionCategory;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.type.OperandTypes;
+import org.apache.calcite.sql.type.ReturnTypes;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.opensearch.test.OpenSearchTestCase;
+
+import java.util.List;
+
+/**
+ * Unit tests for {@link ConvertTzAdapter}. The adapter has three jobs in
+ * priority order: identity short-circuit when both tz operands canonicalize to
+ * the same value, plan-time validation/canonicalization of literal tz operands,
+ * and rewrite to the locally-declared UDF operator otherwise. DST-correct
+ * per-row shifting stays in the Rust UDF since IANA offsets vary per instant.
+ */
+public class ConvertTzAdapterTests extends OpenSearchTestCase {
+
+ private RelDataTypeFactory typeFactory;
+ private RexBuilder rexBuilder;
+ private RelOptCluster cluster;
+
+ @Override
+ public void setUp() throws Exception {
+ super.setUp();
+ typeFactory = new JavaTypeFactoryImpl();
+ rexBuilder = new RexBuilder(typeFactory);
+ HepPlanner planner = new HepPlanner(new HepProgramBuilder().build());
+ cluster = RelOptCluster.create(planner, rexBuilder);
+ }
+
+ private SqlFunction convertTzOp(RelDataType returnType) {
+ return new SqlFunction(
+ "CONVERT_TZ",
+ SqlKind.OTHER_FUNCTION,
+ ReturnTypes.explicit(returnType),
+ null,
+ OperandTypes.ANY_STRING_STRING,
+ SqlFunctionCategory.TIMEDATE
+ );
+ }
+
+ private RexCall buildConvertTz(String fromLit, String toLit) {
+ RelDataType tsType = typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.TIMESTAMP), true);
+ RexNode tsRef = rexBuilder.makeInputRef(tsType, 0);
+ // 2-arg makeLiteral returns a bare RexLiteral; the 3-arg form with a
+ // nullable type wraps in a CAST, which the adapter must then peel back
+ // to inspect the string value. PPL's frontend emits the 2-arg form, so
+ // we match that here.
+ RexNode fromNode = rexBuilder.makeLiteral(fromLit);
+ RexNode toNode = rexBuilder.makeLiteral(toLit);
+ return (RexCall) rexBuilder.makeCall(convertTzOp(tsType), List.of(tsRef, fromNode, toNode));
+ }
+
+ // ── Canonicalization (unit tests on the static helper) ────────────────
+
+ public void testCanonicalizeTzPadsOffsetDigits() {
+ assertEquals("+05:30", ConvertTzAdapter.canonicalizeTz("+5:30"));
+ assertEquals("-08:00", ConvertTzAdapter.canonicalizeTz("-8:00"));
+ assertEquals("+14:00", ConvertTzAdapter.canonicalizeTz("+14:00"));
+ }
+
+ public void testCanonicalizeTzAcceptsIanaNames() {
+ // ZoneId.of passes through canonical ids unchanged.
+ assertEquals("America/New_York", ConvertTzAdapter.canonicalizeTz("America/New_York"));
+ assertEquals("Europe/London", ConvertTzAdapter.canonicalizeTz("Europe/London"));
+ assertEquals("UTC", ConvertTzAdapter.canonicalizeTz("UTC"));
+ }
+
+ public void testCanonicalizeTzRejectsInvalidOffsetBounds() {
+ // Hours > 14 is beyond any real-world zone.
+ IllegalArgumentException ex = expectThrows(IllegalArgumentException.class, () -> ConvertTzAdapter.canonicalizeTz("+15:00"));
+ assertTrue("error must include the bad value: " + ex.getMessage(), ex.getMessage().contains("+15:00"));
+
+ // Minutes > 59 is malformed.
+ expectThrows(IllegalArgumentException.class, () -> ConvertTzAdapter.canonicalizeTz("+05:60"));
+ }
+
+ public void testCanonicalizeTzRejectsUnknownIana() {
+ IllegalArgumentException ex = expectThrows(IllegalArgumentException.class, () -> ConvertTzAdapter.canonicalizeTz("Mars/Olympus"));
+ assertTrue("error must include the bad value for UX: " + ex.getMessage(), ex.getMessage().contains("Mars/Olympus"));
+ }
+
+ // ── adapt() behavior ──────────────────────────────────────────────────
+
+ /**
+ * Identity fold: when both tz literals canonicalize to the same value, the
+ * call reduces to its timestamp operand. No UDF invocation.
+ */
+ public void testAdaptIdentityFoldReturnsTimestampUnchanged() {
+ RexCall original = buildConvertTz("UTC", "UTC");
+ RexNode adapted = new ConvertTzAdapter().adapt(original, List.of(), cluster);
+
+ assertSame("identity fold must return the original timestamp operand", original.getOperands().get(0), adapted);
+ }
+
+ /**
+ * Identity fold must apply *after* canonicalization — `+5:00` and `+05:00`
+ * are the same zone but different strings; the adapter must canonicalize
+ * first, then compare.
+ */
+ public void testAdaptIdentityFoldAppliesAfterCanonicalization() {
+ RexCall original = buildConvertTz("+5:00", "+05:00");
+ RexNode adapted = new ConvertTzAdapter().adapt(original, List.of(), cluster);
+
+ assertSame("identity fold must compare canonical forms", original.getOperands().get(0), adapted);
+ }
+
+ /**
+ * When literals can't be collapsed (IANA pairs, mixed IANA + offset), the
+ * call rewrites to the local UDF operator with canonicalized string
+ * operands. The tz strings passed to the UDF are the canonical form.
+ */
+ public void testAdaptIanaPairRoutesThroughUdfWithCanonicalLiterals() {
+ RexCall original = buildConvertTz("America/New_York", "Europe/London");
+ RexNode adapted = new ConvertTzAdapter().adapt(original, List.of(), cluster);
+
+ assertTrue("adapted node must be a RexCall, got " + adapted.getClass(), adapted instanceof RexCall);
+ RexCall call = (RexCall) adapted;
+ assertSame(
+ "adapted call must target LOCAL_CONVERT_TZ_OP so FunctionMappings.Sig binds",
+ ConvertTzAdapter.LOCAL_CONVERT_TZ_OP,
+ call.getOperator()
+ );
+ assertEquals(3, call.getOperands().size());
+ assertEquals("America/New_York", ((RexLiteral) call.getOperands().get(1)).getValueAs(String.class));
+ assertEquals("Europe/London", ((RexLiteral) call.getOperands().get(2)).getValueAs(String.class));
+ }
+
+ /**
+ * When literal operands need canonicalization (e.g. `+5:00` → `+05:00`),
+ * the UDF-bound call sees the canonical form so the Rust side doesn't need
+ * to do the padding.
+ */
+ public void testAdaptPassesCanonicalizedLiteralsToUdf() {
+ // Pair of distinct-canonical offsets so the fold path doesn't fire.
+ RexCall original = buildConvertTz("+5:00", "+10:00");
+ RexNode adapted = new ConvertTzAdapter().adapt(original, List.of(), cluster);
+
+ assertTrue(adapted instanceof RexCall);
+ RexCall call = (RexCall) adapted;
+ assertSame(ConvertTzAdapter.LOCAL_CONVERT_TZ_OP, call.getOperator());
+ assertEquals("+05:00", ((RexLiteral) call.getOperands().get(1)).getValueAs(String.class));
+ assertEquals("+10:00", ((RexLiteral) call.getOperands().get(2)).getValueAs(String.class));
+ }
+
+ /**
+ * Adapter preserves the original call's return type — matches the
+ * {@code AbstractNameMappingAdapter} regression guard. If the rewritten
+ * call's Calcite-inferred type differs from the original, the enclosing
+ * {@code Project.isValid} compatibleTypes check breaks at fragment
+ * conversion.
+ */
+ public void testAdaptedCallPreservesOriginalReturnType() {
+ RelDataType originalType = typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.TIMESTAMP, 0), true);
+ RexNode tsRef = rexBuilder.makeInputRef(originalType, 0);
+ RexNode fromLit = rexBuilder.makeLiteral("America/New_York");
+ RexNode toLit = rexBuilder.makeLiteral("Europe/London");
+ RexCall original = (RexCall) rexBuilder.makeCall(convertTzOp(originalType), List.of(tsRef, fromLit, toLit));
+ assertEquals(originalType, original.getType());
+
+ RexNode adapted = new ConvertTzAdapter().adapt(original, List.of(), cluster);
+
+ assertEquals(
+ "adapted call's return type must equal the original — otherwise Project.rowType assertion fails",
+ original.getType(),
+ adapted.getType()
+ );
+ }
+
+ /**
+ * Invalid literal tz operand surfaces at plan time as
+ * {@link IllegalArgumentException} with the offending value in the message,
+ * rather than silently producing per-row NULL at runtime.
+ */
+ public void testAdaptInvalidLiteralErrorsAtPlanTime() {
+ RexCall original = buildConvertTz("Mars/Olympus", "UTC");
+ IllegalArgumentException ex = expectThrows(
+ IllegalArgumentException.class,
+ () -> new ConvertTzAdapter().adapt(original, List.of(), cluster)
+ );
+ assertTrue("error must name the offending literal for user UX: " + ex.getMessage(), ex.getMessage().contains("Mars/Olympus"));
+ }
+
+ /**
+ * Column-valued tz operands are not validated at plan time — per-row
+ * values can't be inspected until runtime, so they pass through into the
+ * UDF which handles them leniently (unparseable → NULL row).
+ */
+ public void testAdaptColumnValuedTzOperandsPassThroughToUdf() {
+ RelDataType tsType = typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.TIMESTAMP), true);
+ RelDataType stringType = typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.VARCHAR), true);
+ RexNode tsRef = rexBuilder.makeInputRef(tsType, 0);
+ // Column refs for the tz slots — not literals, so no canonicalization.
+ RexNode fromCol = rexBuilder.makeInputRef(stringType, 1);
+ RexNode toCol = rexBuilder.makeInputRef(stringType, 2);
+ RexCall original = (RexCall) rexBuilder.makeCall(convertTzOp(tsType), List.of(tsRef, fromCol, toCol));
+
+ RexNode adapted = new ConvertTzAdapter().adapt(original, List.of(), cluster);
+
+ assertTrue(adapted instanceof RexCall);
+ RexCall call = (RexCall) adapted;
+ assertSame(ConvertTzAdapter.LOCAL_CONVERT_TZ_OP, call.getOperator());
+ assertSame("column-valued from_tz must pass through unmodified", fromCol, call.getOperands().get(1));
+ assertSame("column-valued to_tz must pass through unmodified", toCol, call.getOperands().get(2));
+ }
+}
diff --git a/sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/UnixTimestampAdapterTests.java b/sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/UnixTimestampAdapterTests.java
new file mode 100644
index 0000000000000..e27216f8ee28d
--- /dev/null
+++ b/sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/UnixTimestampAdapterTests.java
@@ -0,0 +1,112 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.be.datafusion;
+
+import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.hep.HepPlanner;
+import org.apache.calcite.plan.hep.HepProgramBuilder;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.SqlFunction;
+import org.apache.calcite.sql.SqlFunctionCategory;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.type.OperandTypes;
+import org.apache.calcite.sql.type.ReturnTypes;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.opensearch.test.OpenSearchTestCase;
+
+import java.util.List;
+
+/**
+ * Unit tests for {@link UnixTimestampAdapter} — the cat-3a rename adapter that
+ * rewrites PPL's bespoke {@code UNIX_TIMESTAMP} operator to a locally-declared
+ * {@code to_unixtime} {@link SqlFunction} whose {@code FunctionMappings.Sig} we
+ * own. Target name {@code to_unixtime} matches DataFusion's native function; no
+ * UDF registration required on the Rust side.
+ */
+public class UnixTimestampAdapterTests extends OpenSearchTestCase {
+
+ public void testUnixTimestampRewritesToLocalToUnixtimeOperator() {
+ RelDataTypeFactory typeFactory = new JavaTypeFactoryImpl();
+ RexBuilder rexBuilder = new RexBuilder(typeFactory);
+ HepPlanner planner = new HepPlanner(new HepProgramBuilder().build());
+ RelOptCluster cluster = RelOptCluster.create(planner, rexBuilder);
+
+ // Synthesize UNIX_TIMESTAMP(ts) with PPL's return type (DOUBLE_FORCE_NULLABLE).
+ RelDataType tsType = typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.TIMESTAMP), true);
+ RelDataType doubleNullable = typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.DOUBLE), true);
+ SqlFunction unixTimestampOp = new SqlFunction(
+ "UNIX_TIMESTAMP",
+ SqlKind.OTHER_FUNCTION,
+ ReturnTypes.explicit(doubleNullable),
+ null,
+ OperandTypes.ANY,
+ SqlFunctionCategory.TIMEDATE
+ );
+ RexNode tsRef = rexBuilder.makeInputRef(tsType, 0);
+ RexCall original = (RexCall) rexBuilder.makeCall(unixTimestampOp, List.of(tsRef));
+
+ RexNode adapted = new UnixTimestampAdapter().adapt(original, List.of(), cluster);
+
+ assertTrue("adapted node must be a RexCall, got " + adapted.getClass(), adapted instanceof RexCall);
+ RexCall call = (RexCall) adapted;
+ assertSame(
+ "adapted call must target UnixTimestampAdapter.LOCAL_TO_UNIXTIME_OP so the "
+ + "FunctionMappings.Sig in DataFusionFragmentConvertor can bind by reference",
+ UnixTimestampAdapter.LOCAL_TO_UNIXTIME_OP,
+ call.getOperator()
+ );
+ assertEquals("to_unixtime is a pure rename — 1 operand preserved", 1, call.getOperands().size());
+ assertSame("arg 0 must be the original timestamp operand", tsRef, call.getOperands().get(0));
+ }
+
+ /**
+ * Regression guard mirroring {@code YearAdapterTests.testAdaptedCallPreservesOriginalReturnType}.
+ * PPL's {@code UNIX_TIMESTAMP} is typed {@code DOUBLE_FORCE_NULLABLE}; DF's
+ * {@code to_unixtime} is typed {@code Int64}. The adapter must preserve the
+ * original DOUBLE type so the enclosing Project / Filter's cached rowType
+ * doesn't mismatch during fragment conversion. (DataFusion's substrait
+ * consumer re-resolves {@code to_unixtime} by name at plan time and applies
+ * its own coerce_types pass — the Calcite-inferred return type at isthmus
+ * time is purely a plan-validity artifact.)
+ */
+ public void testAdaptedCallPreservesOriginalReturnType() {
+ RelDataTypeFactory typeFactory = new JavaTypeFactoryImpl();
+ RexBuilder rexBuilder = new RexBuilder(typeFactory);
+ HepPlanner planner = new HepPlanner(new HepProgramBuilder().build());
+ RelOptCluster cluster = RelOptCluster.create(planner, rexBuilder);
+
+ RelDataType tsType = typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.TIMESTAMP), true);
+ RelDataType doubleNullable = typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.DOUBLE), true);
+ SqlFunction unixTimestampOp = new SqlFunction(
+ "UNIX_TIMESTAMP",
+ SqlKind.OTHER_FUNCTION,
+ ReturnTypes.explicit(doubleNullable),
+ null,
+ OperandTypes.ANY,
+ SqlFunctionCategory.TIMEDATE
+ );
+ RexNode tsRef = rexBuilder.makeInputRef(tsType, 0);
+ RexCall original = (RexCall) rexBuilder.makeCall(unixTimestampOp, List.of(tsRef));
+ assertEquals(doubleNullable, original.getType());
+
+ RexNode adapted = new UnixTimestampAdapter().adapt(original, List.of(), cluster);
+
+ assertEquals(
+ "adapted call's return type must equal the original — otherwise the enclosing Project.rowType "
+ + "assertion fails during fragment conversion",
+ original.getType(),
+ adapted.getType()
+ );
+ }
+}
diff --git a/sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/YearAdapterTests.java b/sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/YearAdapterTests.java
new file mode 100644
index 0000000000000..a101f74994151
--- /dev/null
+++ b/sandbox/plugins/analytics-backend-datafusion/src/test/java/org/opensearch/be/datafusion/YearAdapterTests.java
@@ -0,0 +1,116 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.be.datafusion;
+
+import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.hep.HepPlanner;
+import org.apache.calcite.plan.hep.HepProgramBuilder;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.SqlFunction;
+import org.apache.calcite.sql.SqlFunctionCategory;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.fun.SqlLibraryOperators;
+import org.apache.calcite.sql.type.OperandTypes;
+import org.apache.calcite.sql.type.ReturnTypes;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.opensearch.analytics.spi.AbstractNameMappingAdapter;
+import org.opensearch.test.OpenSearchTestCase;
+
+import java.util.List;
+
+/**
+ * Unit tests for {@link YearAdapter} exercising the reusable rename +
+ * literal-arg injection adapter pattern via {@link AbstractNameMappingAdapter}.
+ */
+public class YearAdapterTests extends OpenSearchTestCase {
+
+ public void testYearRewritesToDatePartWithYearLiteral() {
+ RelDataTypeFactory typeFactory = new JavaTypeFactoryImpl();
+ RexBuilder rexBuilder = new RexBuilder(typeFactory);
+ HepPlanner planner = new HepPlanner(new HepProgramBuilder().build());
+ RelOptCluster cluster = RelOptCluster.create(planner, rexBuilder);
+
+ // Synthesize YEAR(ts) — a one-arg Calcite call of our own SqlFunction
+ // so the test doesn't depend on any specific builtin.
+ RelDataType tsType = typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.TIMESTAMP), true);
+ SqlFunction yearOp = new SqlFunction(
+ "YEAR",
+ SqlKind.OTHER_FUNCTION,
+ ReturnTypes.explicit(typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.BIGINT), true)),
+ null,
+ OperandTypes.ANY,
+ SqlFunctionCategory.TIMEDATE
+ );
+ RexNode tsRef = rexBuilder.makeInputRef(tsType, 0);
+ RexCall original = (RexCall) rexBuilder.makeCall(yearOp, List.of(tsRef));
+
+ RexNode adapted = new YearAdapter().adapt(original, List.of(), cluster);
+
+ assertTrue("adapted node must be a RexCall, got " + adapted.getClass(), adapted instanceof RexCall);
+ RexCall call = (RexCall) adapted;
+ assertEquals("adapted call must target DATE_PART", SqlLibraryOperators.DATE_PART, call.getOperator());
+ assertEquals("date_part(unit, value) must have 2 operands after year-literal prepend", 2, call.getOperands().size());
+ assertTrue(
+ "arg 0 must be a string literal, got " + call.getOperands().get(0).getClass(),
+ call.getOperands().get(0) instanceof RexLiteral
+ );
+ RexLiteral unitLit = (RexLiteral) call.getOperands().get(0);
+ assertEquals("year", unitLit.getValueAs(String.class));
+ assertSame("arg 1 must be the original operand", tsRef, call.getOperands().get(1));
+ }
+
+ /**
+ * The adapter MUST preserve the Calcite {@link RelDataType} of the original call.
+ * Otherwise the enclosing Project's cached {@code rowType} (derived from the pre-
+ * adaptation expression) mismatches the adapted expression's type, tripping
+ * {@code Project.isValid}'s {@code RexUtil.compatibleTypes} assertion during
+ * fragment conversion. Regression guard for the PR10 IT hang where
+ * {@code DATE_PART} produced a different Calcite-inferred type than {@code YEAR}.
+ */
+ public void testAdaptedCallPreservesOriginalReturnType() {
+ RelDataTypeFactory typeFactory = new JavaTypeFactoryImpl();
+ RexBuilder rexBuilder = new RexBuilder(typeFactory);
+ HepPlanner planner = new HepPlanner(new HepProgramBuilder().build());
+ RelOptCluster cluster = RelOptCluster.create(planner, rexBuilder);
+
+ RelDataType tsType = typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.TIMESTAMP), true);
+ // PPL's YEAR operator is registered with INTEGER_FORCE_NULLABLE — distinct
+ // from Calcite's SqlLibraryOperators.DATE_PART (which returns BIGINT via
+ // SqlExtractFunction). If the adapter didn't clone with the original's type,
+ // the Project's cached rowType (derived from INTEGER) would clash with the
+ // adapted DATE_PART's inferred BIGINT, tripping Project.isValid.
+ RelDataType integerNullable = typeFactory.createTypeWithNullability(typeFactory.createSqlType(SqlTypeName.INTEGER), true);
+ SqlFunction yearOp = new SqlFunction(
+ "YEAR",
+ SqlKind.OTHER_FUNCTION,
+ ReturnTypes.explicit(integerNullable),
+ null,
+ OperandTypes.ANY,
+ SqlFunctionCategory.TIMEDATE
+ );
+ RexNode tsRef = rexBuilder.makeInputRef(tsType, 0);
+ RexCall original = (RexCall) rexBuilder.makeCall(yearOp, List.of(tsRef));
+ assertEquals(integerNullable, original.getType());
+
+ RexNode adapted = new YearAdapter().adapt(original, List.of(), cluster);
+
+ assertEquals(
+ "adapted call's return type must equal the original call's return type, "
+ + "otherwise the enclosing Project.rowType assertion fails in fragment conversion",
+ original.getType(),
+ adapted.getType()
+ );
+ }
+}