fix(sql): fix alter table dedup enable syntax (#3643)

core/src/main/java/io/questdb/cairo/DedupColumnCommitAddresses.java

@@ -45,7 +45,7 @@ public class DedupColumnCommitAddresses implements Closeable {
     private static final long NULL_VAL_256 = RESERVED3 + 8L;
     private static final int RECORD_BYTES = (int) (NULL_VAL_256 + 32L);
     // The data structure in above offsets has to match dedup_column struct in dedup.cpp
-    
+
     private PagedDirectLongList addresses;
     private int columnCount;
 
@@ -127,8 +127,10 @@ public class DedupColumnCommitAddresses implements Closeable {
             }
             int longsPerBlock = RECORD_BYTES / Long.BYTES;
             addresses.setBlockSize(dedupColumnCount * longsPerBlock);
-            this.columnCount = dedupColumnCount;
+        } else if (dedupColumnCount == 0) {
+            clear();
         }
+        this.columnCount = dedupColumnCount;
     }
 
     static {

core/src/main/java/io/questdb/cairo/TableWriter.java

@@ -7569,8 +7569,7 @@ public class TableWriter implements TableWriterAPI, MetadataService, Closeable {
             }
             dedupColumnCommitAddresses.setDedupColumnCount(columnsIndexes.size() - 1);
         } else {
-            if (dedupColumnCommitAddresses == null) {
-                dedupColumnCommitAddresses.clear();
+            if (dedupColumnCommitAddresses != null) {
                 dedupColumnCommitAddresses.setDedupColumnCount(0);
             }
         }

core/src/main/java/io/questdb/griffin/SqlCompilerImpl.java

@@ -876,6 +876,13 @@ public class SqlCompilerImpl implements SqlCompiler, Closeable {
 
         boolean tsIncludedInDedupColumns = false;
 
+        // ALTER TABLE abc DEDUP <ENABLE> UPSERT KEYS(a, b)
+        // ENABLE word is not mandatory to be compatible v7.3
+        // where it was omitted from the syntax
+        if (tok != null && isEnableKeyword(tok)) {
+            tok = SqlUtil.fetchNext(lexer);
+        }
+
         if (tok == null || !isUpsertKeyword(tok)) {
             throw SqlException.position(lexer.lastTokenPosition()).put("expected 'upsert'");
         }

core/src/main/java/io/questdb/griffin/SqlKeywords.java

@@ -26,6 +26,7 @@ package io.questdb.griffin;
 
 import io.questdb.std.Chars;
 import io.questdb.std.LowerCaseCharSequenceHashSet;
+import org.jetbrains.annotations.NotNull;
 
 public class SqlKeywords {
     public static final int CASE_KEYWORD_LENGTH = 4;
@@ -596,6 +597,20 @@ public class SqlKeywords {
                 && (tok.charAt(i) | 32) == 'p';
     }
 
+    public static boolean isEnableKeyword(@NotNull CharSequence tok) {
+        if (tok.length() != 6) {
+            return false;
+        }
+
+        int i = 0;
+        return (tok.charAt(i++) | 32) == 'e'
+                && (tok.charAt(i++) | 32) == 'n'
+                && (tok.charAt(i++) | 32) == 'a'
+                && (tok.charAt(i++) | 32) == 'b'
+                && (tok.charAt(i++) | 32) == 'l'
+                && (tok.charAt(i) | 32) == 'e';
+    }
+
     public static boolean isEndKeyword(CharSequence tok) {
         if (tok.length() != 3) {
             return false;

core/src/main/java/io/questdb/griffin/engine/functions/catalogue/TableListFunctionFactory.java

@@ -179,7 +179,7 @@ public class TableListFunctionFactory implements FunctionFactory {
                     }
                     if (col == DEDUP_NAME_COLUMN) {
                         int timestampIndex = tableReaderMetadata.getTimestampIndex();
-                        return timestampIndex > 0 && tableReaderMetadata.isWalEnabled() && tableReaderMetadata.isDedupKey(timestampIndex);
+                        return timestampIndex >= 0 && tableReaderMetadata.isWalEnabled() && tableReaderMetadata.isDedupKey(timestampIndex);
                     }
                     return false;
                 }
@@ -194,7 +194,7 @@ public class TableListFunctionFactory implements FunctionFactory {
 
                 @Override
                 public long getLong(int col) {
-                        return o3MaxLag;
+                    return o3MaxLag;
                 }
 
                 @Override

core/src/test/java/io/questdb/test/griffin/CreateTableDedupTest.java

@@ -347,6 +347,50 @@ public class CreateTableDedupTest extends AbstractCairoTest {
         });
     }
 
+    @Test
+    public void testEnableDedup() throws Exception {
+        String tableName = testName.getMethodName();
+        ddl(
+                "create table " + tableName +
+                        " (ts TIMESTAMP, x long, s symbol) timestamp(ts)" +
+                        " PARTITION BY DAY WAL DEDUP UPSERT KEYS(ts,s)"
+        );
+        assertSql(
+                "name\tdedup\n" +
+                        "testEnableDedup\ttrue\n",
+                "select name, dedup from tables() where name ='" + tableName + "'"
+        );
+        assertSql(
+                "column\ttype\tindexed\tindexBlockCapacity\tsymbolCached\tsymbolCapacity\tdesignated\tupsertKey\n" +
+                        "ts\tTIMESTAMP\tfalse\t0\tfalse\t0\ttrue\ttrue\n" +
+                        "x\tLONG\tfalse\t0\tfalse\t0\tfalse\tfalse\n" +
+                        "s\tSYMBOL\tfalse\t256\ttrue\t128\tfalse\ttrue\n",
+                "show columns from '" + tableName + "'"
+        );
+        compile("alter table " + tableName + " dedup disable");
+        drainWalQueue();
+        assertSql(
+                "name\tdedup\n" +
+                        "testEnableDedup\tfalse\n",
+                "select name, dedup from tables() where name ='" + tableName + "'"
+        );
+
+        compile("alter table " + tableName + " dedup enable upsert keys(ts)");
+        drainWalQueue();
+        assertSql(
+                "name\tdedup\n" +
+                        "testEnableDedup\ttrue\n",
+                "select name, dedup from tables() where name ='" + tableName + "'"
+        );
+        assertSql(
+                "column\ttype\tindexed\tindexBlockCapacity\tsymbolCached\tsymbolCapacity\tdesignated\tupsertKey\n" +
+                        "ts\tTIMESTAMP\tfalse\t0\tfalse\t0\ttrue\ttrue\n" +
+                        "x\tLONG\tfalse\t0\tfalse\t0\tfalse\tfalse\n" +
+                        "s\tSYMBOL\tfalse\t256\ttrue\t128\tfalse\tfalse\n",
+                "show columns from '" + tableName + "'"
+        );
+    }
+
     @Test
     public void testEnableDedupDroppedColumnColumnConcurrently() throws Exception {
         assertMemoryLeak(() -> {

core/src/test/java/io/questdb/test/griffin/wal/DedupInsertFuzzTest.java

@@ -55,7 +55,9 @@ public class DedupInsertFuzzTest extends AbstractFuzzTest {
     public void testDedupWithRandomShiftAndStep() throws Exception {
         assertMemoryLeak(() -> {
             String tableName = testName.getMethodName();
-            createEmptyTable(tableName, "DEDUP upsert keys(ts)");
+            createEmptyTable(tableName, "DEDUP upsert keys(ts, commit)");
+            compile("alter table " + tableName + " dedup disable");
+            compile("alter table " + tableName + " dedup enable upsert keys(ts)");
 
             ObjList<FuzzTransaction> transactions = new ObjList<>();
             Rnd rnd = generateRandom(LOG);