feldera · mihaibudiu · Jul 1, 2026 · Jun 29, 2026
diff --git a/crates/sqllib/src/timestamp.rs b/crates/sqllib/src/timestamp.rs
@@ -2866,13 +2866,51 @@ some_polymorphic_function1!(extract_second, Date, Date, i64);
 some_polymorphic_function1!(extract_minute, Date, Date, i64);
 some_polymorphic_function1!(extract_hour, Date, Date, i64);
 
+#[doc(hidden)]
+pub fn datediff_hour_Date_Date(left: Date, right: Date) -> i32 {
+    (right.days() - left.days()) * 24
+}
+
+some_polymorphic_function2!(datediff_hour, Date, Date, Date, Date, i32);
+
 #[doc(hidden)]
 pub fn datediff_day_Date_Date(left: Date, right: Date) -> i32 {
-    left.days() - right.days()
+    right.days() - left.days()
 }
 
 some_polymorphic_function2!(datediff_day, Date, Date, Date, Date, i32);
 
+#[doc(hidden)]
+pub fn datediff_week_Date_Date(left: Date, right: Date) -> i32 {
+    (right.days() - left.days()) / 7
+}
+
+some_polymorphic_function2!(datediff_week, Date, Date, Date, Date, i32);
+
+#[doc(hidden)]
+pub fn datediff_month_Date_Date(left: Date, right: Date) -> i32 {
+    let interval = minus_LongInterval_Date_Date__(right, left);
+    interval.months()
+}
+
+some_polymorphic_function2!(datediff_month, Date, Date, Date, Date, i32);
+
+#[doc(hidden)]
+pub fn datediff_quarter_Date_Date(left: Date, right: Date) -> i32 {
+    datediff_year_Date_Date(left, right) * 4
+        + (extract_quarter_Date(right) - extract_quarter_Date(left)) as i32
+}
+
+some_polymorphic_function2!(datediff_quarter, Date, Date, Date, Date, i32);
+
+#[doc(hidden)]
+pub fn datediff_year_Date_Date(left: Date, right: Date) -> i32 {
+    let interval = minus_LongInterval_Date_Date__(right, left);
+    interval.years()
+}
+
+some_polymorphic_function2!(datediff_year, Date, Date, Date, Date, i32);
+
 #[doc(hidden)]
 pub fn format_date__(format: SqlString, date: Date) -> SqlString {
     SqlString::from(date.to_dateTime().format(format.str()).to_string())

diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/codegen/includes/parserImpls.ftl b/sql-to-dbsp-compiler/SQL-compiler/src/main/codegen/includes/parserImpls.ftl
@@ -36,7 +36,7 @@ SqlNode DateaddFunctionCall() :
 {
     (   <DATE_PART>  { op = SqlLibraryOperators.DATE_PART; }
     |   <DATEADD> { op = SqlLibraryOperators.DATEADD; }
-    |   <DATEDIFF> { op = SqlStdOperatorTable.TIMESTAMP_DIFF; }
+    |   <DATEDIFF> { op = SqlLibraryOperators.DATEDIFF; }
     |   <DATEPART>  { op = SqlLibraryOperators.DATEPART; }
     )
     { s = span(); }

diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/CompilerMain.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/CompilerMain.java
@@ -175,7 +175,7 @@ CompilerMessages run() {
             return compiler.messages;
         // The following runs all compilation stages
         DBSPCircuit circuit = compiler.getFinalCircuit(false);
-        if (options.ioOptions.anonymize)
+        if (options.ioOptions.anonymize || options.ioOptions.format)
             // The front-end has already emitted the output, we are done.
             return compiler.messages;
         if (compiler.hasErrors())

diff --git a/...sp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/CompilerOptions.java b/...sp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/CompilerOptions.java
@@ -159,6 +159,9 @@ public String diff(CompilerOptions other) {
     public static class IO implements IDiff<IO>, IValidate {
         @Parameter(names = "--anonymize", description = "Produce in the output file an anonymized version of the input program")
         public boolean anonymize = false;
+        @Parameter(names = "--format",
+                description = "Output the SQL program reformatted")
+        public boolean format = false;
         @DynamicParameter(names = "-T",
                 description = "Specify logging level for a class (can be repeated)")
         public Map<String, String> loggingLevel = new HashMap<>();
@@ -278,6 +281,7 @@ public String toString() {
                     ",\n\tnoRust=" + this.noRust +
                     ",\n\toutputFile=" + Utilities.singleQuote(this.outputFile) +
                     ",\n\tquiet=" + this.quiet +
+                    ",\n\tformat=" + this.format +
                     ",\n\truntime=" + Utilities.singleQuote(this.runtimePath) +
                     ",\n\ttrimInputs=" + this.trimInputs +
                     ",\n\tverbosity=" + this.verbosity +

diff --git a/...-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/DBSPCompiler.java b/...-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/DBSPCompiler.java
@@ -39,6 +39,9 @@
 import org.apache.calcite.sql.dialect.CalciteSqlDialect;
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.calcite.sql.pretty.SqlPrettyWriter;
+import org.dbsp.sqlCompiler.compiler.frontend.SqlComment;
+import org.dbsp.sqlCompiler.compiler.frontend.SqlCommentParser;
+import org.dbsp.sqlCompiler.compiler.frontend.SqlPrettyPrinter;
 import org.apache.calcite.sql.util.SqlOperatorTables;
 import org.apache.calcite.sql.validate.SqlUserDefinedAggFunction;
 import org.dbsp.sqlCompiler.circuit.DBSPCircuit;
@@ -98,7 +101,7 @@
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
-import java.util.Locale;
+import java.util.Objects;
 import java.util.Map;
 import java.util.regex.Pattern;
 
@@ -532,7 +535,7 @@ boolean validateCreateIndex(CreateIndexStatement statement) {
         return true;
     }
 
-    void renameIdentifiers(List<ParsedStatement> statements) {
+    void emitSql(List<ParsedStatement> statements) {
         final PrintStream outputStream;
         try {
             @Nullable String outputFile = this.options.ioOptions.outputFile;
@@ -547,20 +550,36 @@ void renameIdentifiers(List<ParsedStatement> statements) {
             return;
         }
 
+        List<SqlComment> comments = this.options.ioOptions.format
+                ? SqlCommentParser.parse(this.sources.getWholeProgram())
+                : new ArrayList<>();
+        IndentStream indentStream = new IndentStream(outputStream);
+        SqlPrettyPrinter printer = new SqlPrettyPrinter(indentStream, comments);
+
         RenameIdentifiers renamer = new RenameIdentifiers();
-        for (ParsedStatement stat: statements) {
+        for (ParsedStatement stat : statements) {
             if (stat.visible()) {
-                SqlNode renamed = renamer.visitNode(stat.statement());
-                Utilities.enforce(renamed != null);
-                final SqlWriter sqlWriter = new SqlPrettyWriter(
-                        SqlWriterConfig.of()
-                                .withDialect(CalciteSqlDialect.DEFAULT)
-                                .withQuoteAllIdentifiers(false)
-                );
-                renamed.unparse(sqlWriter, 0, 0);
-                outputStream.println(sqlWriter + ";");
+                SqlNode node = stat.statement();
+                if (this.options.ioOptions.anonymize)
+                    node = Objects.requireNonNull(renamer.visitNode(node));
+                Utilities.enforce(node != null);
+                if (this.options.ioOptions.format) {
+                    printer.print(node);
+                    indentStream.append(";").newline();
+                } else {
+                    final SqlWriter sqlWriter = new SqlPrettyWriter(
+                            SqlWriterConfig.of()
+                                    .withDialect(CalciteSqlDialect.DEFAULT)
+                                    .withQuoteAllIdentifiers(false)
+                    );
+                    node.unparse(sqlWriter, 0, 0);
+                    outputStream.println(sqlWriter + ";");
+                }
             }
         }
+        if (this.options.ioOptions.format) {
+            printer.flushRemainingComments();
+        }
         outputStream.close();
     }
 
@@ -569,8 +588,8 @@ void renameIdentifiers(List<ParsedStatement> statements) {
         if (this.hasErrors())
             return null;
 
-        if (this.options.ioOptions.anonymize) {
-            this.renameIdentifiers(parsed);
+        if (this.options.ioOptions.anonymize || this.options.ioOptions.format) {
+            this.emitSql(parsed);
             return null;
         }
 
@@ -749,7 +768,8 @@ void renameIdentifiers(List<ParsedStatement> statements) {
         return null;
     }
 
-    static final Pattern ITEM_ERROR = Pattern.compile("Cannot apply 'ITEM' to arguments of type 'ITEM\\(([^,]+), ([^']+)\\)'(.*)", Pattern.DOTALL);
+    static final Pattern ITEM_ERROR = Pattern.compile(
+            "Cannot apply 'ITEM' to arguments of type 'ITEM\\(([^,]+), ([^']+)\\)'(.*)", Pattern.DOTALL);
 
     static SourcePositionRange getRange(CalciteContextException e) {
         return new SourcePositionRange(
@@ -769,23 +789,6 @@ private CompilationError improveErrorMessage(CalciteContextException e) {
                 String newMessage = "Cannot apply indexing to arguments of type " + source + "[" + index + "]" + tail;
                 return new CompilationError(newMessage, getRange(e));
             }
-            if (message.contains("'TIMESTAMPDIFF'.")) {
-                // The Calcite parser replaces DATEDIFF with TIMESTAMPDIFF
-                // Try to figure out whether this is what happened and rewrite it.
-                // This heuristic may fail... but will work in general.
-                SourcePositionRange range = getRange(e);
-                String fragment = this.sources.getFragment(range, false);
-                String lower = fragment.toLowerCase(Locale.ENGLISH);
-                if (lower.contains("datediff") && !lower.contains("timestampdiff")) {
-                    message = message.replace("TIMESTAMPDIFF", "DATEDIFF");
-                    return new CompilationError(message, range);
-                }
-            }
-            String newMessage = message.replace(":PEEK_NO_EXPAND", "");
-            newMessage = newMessage.replace("RECORDTYPE", "ROW");
-            if (!newMessage.equals(message)) {
-                return new CompilationError(newMessage, getRange(e));
-            }
         }
         return new CompilationError(e);
     }

diff --git a/...SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/ExpressionCompiler.java b/...SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/ExpressionCompiler.java
@@ -107,6 +107,7 @@
 import org.dbsp.sqlCompiler.ir.type.primitive.DBSPTypeBaseType;
 import org.dbsp.sqlCompiler.ir.type.primitive.DBSPTypeDate;
 import org.dbsp.sqlCompiler.ir.type.primitive.DBSPTypeFP;
+import org.dbsp.sqlCompiler.ir.type.primitive.DBSPTypeKeyword;
 import org.dbsp.sqlCompiler.ir.type.primitive.DBSPTypeTime;
 import org.dbsp.sqlCompiler.ir.type.primitive.DBSPTypeTimestampTz;
 import org.dbsp.sqlCompiler.ir.type.primitive.DBSPTypeVariant;
@@ -142,6 +143,7 @@
 import java.nio.charset.Charset;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Locale;
 import java.util.Objects;
 import java.util.UUID;
 
@@ -818,6 +820,18 @@ void ensureInteger(CalciteObject node, List<DBSPExpression> ops, int argument) {
             ops.set(argument, arg.cast(node, expected, DBSPCastExpression.CastType.SqlUnsafe));
     }
 
+    /** Ensure that a specific operand is interpreted as a KEYWORD */
+    void ensureKeyword(CalciteObject node, List<DBSPExpression> ops, int argument) {
+        DBSPExpression arg = ops.get(argument);
+        if (arg.getType().sameType(DBSPTypeKeyword.INSTANCE))
+            return;
+        if (arg.is(DBSPStringLiteral.class)) {
+            ops.set(argument, new DBSPKeywordLiteral(node, Objects.requireNonNull(arg.to(DBSPStringLiteral.class).value)));
+            return;
+        }
+        throw new CompilationError("Unknown keyword", node);
+    }
+
     @SuppressWarnings("SameParameterValue")
     void nullLiteralToNullArray(List<DBSPExpression> ops, int arg) {
         if (ops.get(arg).is(DBSPNullLiteral.class)) {
@@ -1934,6 +1948,24 @@ else if (arg0Type.is(DBSPTypeMap.class))
                 this.ensureString(ops, 2);
                 return compileKeywordFunction(call, node, null, type, ops, 0, 3);
             case TIMESTAMP_DIFF:
+                // Also called for DATEDIFF
+                this.ensureKeyword(node, ops, 0);
+                DBSPKeywordLiteral lit = ops.get(0).to(DBSPKeywordLiteral.class);
+                if (call.op.getName().toLowerCase(Locale.ENGLISH).equals("datediff")) {
+                    switch (lit.keyword) {
+                        // These units would overflow in 32 bits
+                        case "minute":
+                        case "second":
+                        case "millisecond":
+                        case "microsecond":
+                        case "nanosecond":
+                            throw new CompilationError("Use TIMESTAMPDIFF instead of DATEDIFF for " +
+                                    lit.keyword.toUpperCase(Locale.ENGLISH) +
+                                    "\n(TIMESTAMPDIFF result is BIGINT)", node);
+                        default:
+                            break;
+                    }
+                }
                 return compileKeywordFunction(call, node, null, type, ops, 0, 3);
             case TUMBLE: {
                 validateArgCount(node, operationName, ops.size(), 2, 3);

diff --git a/...ompiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/SqlComment.java b/...ompiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/SqlComment.java
@@ -0,0 +1,32 @@
+package org.dbsp.sqlCompiler.compiler.frontend;
+
+import org.apache.calcite.sql.parser.SqlParserPos;
+
+/** A SQL comment extracted from the original source text, with its source position. */
+public class SqlComment {
+    public enum Kind {
+        /** {@code -- ...} to end of line (SQL standard). */
+        LINE_DASH,
+        /** {@code // ...} to end of line (C++ style). */
+        LINE_SLASH,
+        /** {@code /* ... *\/} possibly spanning multiple lines. */
+        BLOCK
+    }
+
+    public final Kind kind;
+    /** Full comment text, including delimiters (e.g. {@code -- foo\n} or {@code /* bar *\/}). */
+    public final String text;
+    /** Position of the first character of the comment in the original source. */
+    public final SqlParserPos pos;
+
+    public SqlComment(Kind kind, String text, SqlParserPos pos) {
+        this.kind = kind;
+        this.text = text;
+        this.pos = pos;
+    }
+
+    @Override
+    public String toString() {
+        return pos + ": " + text.trim();
+    }
+}