[ASTERIXDB-3702][RT][STO] LSM Sampling

- user model changes: yes
- storage format changes: yes
- interface changes: yes

Details:
Introduced sample cursor to collect random
objects from the leaf, in an attempt to improve
the sampling performance by ditching the full scan.

Ext-ref: MB-68399
Change-Id: If5c5b7eac5199b85fc40b722c87ae0bbb73eecbd
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/20959
Reviewed-by: Murtadha Hubail <mhubail@apache.org>
Tested-by: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Reviewed-by: Ritik Raj <ritik.raj@couchbase.com>
Integration-Tests: Jenkins <jenkins@fulliautomatix.ics.uci.edu>

Author: toad-sage

asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/am/InvertedIndexAccessMethod.java

@@ -24,6 +24,7 @@
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 
@@ -588,23 +589,42 @@ public boolean applyJoinPlanTransformation(List<Mutable<ILogicalOperator>> after
         context.computeAndSetTypeEnvironmentForOperator(topSelect);
         ILogicalOperator topOp = topSelect;
 
+        // Maps each index-subtree variable consumed by the panic UnionAll to the fresh variable
+        // the UnionAll produces in its place. Used below the equi-join to rewire operators above
+        // the join (afterJoinRefs) off the now-consumed input variables. Empty when no panic plan.
+        Map<LogicalVariable, LogicalVariable> indexVarToOutVar = new LinkedHashMap<>();
         // Hook up the indexed-nested loop join path with the "panic" (non indexed) nested-loop join path by putting a union all on top.
         if (panicJoinRef != null) {
             LogicalVariable inputSearchVar = getInputSearchVar(optFuncExpr, indexSubTree);
             indexSubTreeLiveVars.addAll(originalSubTreePKs);
             indexSubTreeLiveVars.add(inputSearchVar);
             List<LogicalVariable> panicPlanLiveVars = new ArrayList<>();
             VariableUtilities.getLiveVariables(panicJoinRef.getValue(), panicPlanLiveVars);
-            // Create variable mapping for union all operator.
+            // Create variable mapping for the union all operator. Each output variable must be a
+            // FRESH variable so produced variables are disjoint from used (input) variables — the
+            // SSA invariant checked by PlanStructureVerifier, which reusing the input var as the
+            // output violates. indexSubTreeLiveVars can contain duplicates (getLiveVariables above
+            // already returns the PKs, which are then added again, and inputSearchVar may already
+            // be live), so dedupe first: exactly one fresh output per distinct input. Otherwise the
+            // same input maps to two outputs and downstream references to the un-chosen one dangle.
             List<Triple<LogicalVariable, LogicalVariable, LogicalVariable>> varMap = new ArrayList<>();
-            for (int i = 0; i < indexSubTreeLiveVars.size(); i++) {
-                LogicalVariable indexSubTreeVar = indexSubTreeLiveVars.get(i);
+            for (LogicalVariable indexSubTreeVar : new LinkedHashSet<>(indexSubTreeLiveVars)) {
                 LogicalVariable panicPlanVar = panicVarMap.get(indexSubTreeVar);
                 if (panicPlanVar == null) {
                     panicPlanVar = indexSubTreeVar;
                 }
+                LogicalVariable outVar = context.newVar();
+                indexVarToOutVar.put(indexSubTreeVar, outVar);
                 varMap.add(new Triple<LogicalVariable, LogicalVariable, LogicalVariable>(indexSubTreeVar, panicPlanVar,
-                        indexSubTreeVar));
+                        outVar));
+            }
+            // Remap originalSubTreePKs to the UnionAll output vars so the downstream equi-join
+            // condition references the post-UnionAll variables.
+            for (int i = 0; i < originalSubTreePKs.size(); i++) {
+                LogicalVariable outVar = indexVarToOutVar.get(originalSubTreePKs.get(i));
+                if (outVar != null) {
+                    originalSubTreePKs.set(i, outVar);
+                }
             }
             UnionAllOperator unionAllOp = new UnionAllOperator(varMap);
             unionAllOp.setSourceLocation(topOp.getSourceLocation());
@@ -627,6 +647,18 @@ public boolean applyJoinPlanTransformation(List<Mutable<ILogicalOperator>> after
         joinRef.setValue(topEqJoin);
         context.computeAndSetTypeEnvironmentForOperator(topEqJoin);
 
+        // When a panic UnionAll was inserted, the index-subtree variables it consumed are no longer
+        // live above the join — the UnionAll now produces fresh variables in their place. Rewire the
+        // operators above the join (afterJoinRefs) from the old variables to the fresh UnionAll
+        // outputs so their references don't dangle (undefined-used-variable). Substitute in each
+        // operator (self only); afterJoinRefs already enumerates the full above-join chain.
+        if (!indexVarToOutVar.isEmpty()) {
+            for (Mutable<ILogicalOperator> aboveJoinRef : afterJoinRefs) {
+                VariableUtilities.substituteVariables(aboveJoinRef.getValue(), indexVarToOutVar, context);
+                context.computeAndSetTypeEnvironmentForOperator(aboveJoinRef.getValue());
+            }
+        }
+
         return true;
     }
 
@@ -839,9 +871,11 @@ private void createIsFilterableSelectOps(ILogicalOperator inputOp, LogicalVariab
         isFilterableSelectOp.setExecutionMode(ExecutionMode.LOCAL);
         context.computeAndSetTypeEnvironmentForOperator(isFilterableSelectOp);
 
-        // Select operator for removing tuples that are filterable.
+        // Select operator for removing tuples that are filterable. Clone the expression: the same
+        // isFilterableExpr instance is already owned by isFilterableSelectOp above, and an operator
+        // graph must not share expression object instances across operators (PlanStructureVerifier).
         List<Mutable<ILogicalExpression>> isNotFilterableArgs = new ArrayList<Mutable<ILogicalExpression>>();
-        isNotFilterableArgs.add(new MutableObject<ILogicalExpression>(isFilterableExpr));
+        isNotFilterableArgs.add(new MutableObject<ILogicalExpression>(isFilterableExpr.cloneExpression()));
         ScalarFunctionCallExpression isNotFilterableExpr = new ScalarFunctionCallExpression(
                 FunctionUtil.getFunctionInfo(BuiltinFunctions.NOT), isNotFilterableArgs);
         isNotFilterableExpr.setSourceLocation(sourceLoc);

asterixdb/asterix-app/src/main/java/org/apache/asterix/app/function/DumpIndexReader.java

@@ -154,7 +154,11 @@ private void printField(StringBuilder sb, IAObject field) {
             case STRING:
                 JSONUtil.quoteAndEscape(recordBuilder, ((AString) field).getStringValue());
                 break;
+            case NULL:
+                sb.append("null");
+                break;
             case MISSING:
+                // Should not reach here - MISSING fields filtered in printObject/buildJsonRecord
                 break;
             default:
                 sb.append(field);
@@ -165,14 +169,20 @@ private void printObject(StringBuilder sb, ARecord record) {
         sb.append("{ ");
         int num = record.numberOfFields();
         ARecordType type = record.getType();
+        boolean firstField = true;
         for (int i = 0; i < num; i++) {
-            if (i > 0) {
+            IAObject value = record.getValueByPos(i);
+            // Skip MISSING fields to avoid malformed JSON like "field":,
+            if (value.getType().getTypeTag() == ATypeTag.MISSING) {
+                continue;
+            }
+            if (!firstField) {
                 sb.append(", ");
             }
-            IAObject value = record.getValueByPos(i);
             JSONUtil.quoteAndEscape(sb, type.getFieldNames()[i]);
             sb.append(": ");
             printField(sb, value);
+            firstField = false;
         }
         sb.append(" }");
     }

asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java

@@ -5235,10 +5235,34 @@ protected void doAnalyzeDataset(MetadataProvider metadataProvider, AnalyzeStatem
             InternalDatasetDetails dsDetails = (InternalDatasetDetails) ds.getDatasetDetails();
             int sampleCardinalityTarget = stmtAnalyze.getSampleSize();
             long sampleSeed = stmtAnalyze.getOrCreateSampleSeed();
+            Index.SampleIndexDetails.SampleMethod sampleMethod = stmtAnalyze.getSampleMethod();
+
+            // Random sampling requires every primary-index disk component to carry the sampling metadata (theta
+            // sketch + max leaf tuple count). Components predating the feature acquire it only once merged; a freshly
+            // flushed component always carries it, so probing the existing disk components is sufficient. If any
+            // component is missing the metadata, an unbiased random sample is impossible, so fall back to full scan.
+            if (sampleMethod != Index.SampleIndexDetails.SampleMethod.FULL_SCAN) {
+                JobSpecification probeSpec = DatasetUtil.buildSamplingMetadataProbeJobSpec(ds, metadataProvider);
+                MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
+                bActiveTxn = false;
+                List<IOperatorStats> probeStats = runJob(hcc, probeSpec, jobFlags,
+                        Collections.singletonList(DatasetUtil.SAMPLING_METADATA_PROBE_OPERATOR_NAME));
+                // No stats => could not determine; be conservative and force a full scan.
+                long componentsMissingMetadata =
+                        probeStats == null || probeStats.isEmpty() ? -1 : probeStats.get(0).getTupleCounter().get();
+                if (componentsMissingMetadata != 0) {
+                    LOGGER.warn("Dataset '{}' has {} primary-index disk component(s) missing random-sampling "
+                            + "metadata; forcing full scan.", datasetName, componentsMissingMetadata);
+                    sampleMethod = Index.SampleIndexDetails.SampleMethod.FULL_SCAN;
+                }
+                mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
+                bActiveTxn = true;
+                metadataProvider.setMetadataTxnContext(mdTxnCtx);
+            }
 
             Index.SampleIndexDetails newIndexDetailsPendingAdd = new Index.SampleIndexDetails(dsDetails.getPrimaryKey(),
                     dsDetails.getKeySourceIndicator(), dsDetails.getPrimaryKeyType(), sampleCardinalityTarget, 0, 0,
-                    sampleSeed, Collections.emptyMap());
+                    sampleSeed, sampleMethod, Collections.emptyMap());
             newIndexPendingAdd = new Index(databaseName, dataverseName, datasetName, newIndexName, sampleIndexType,
                     newIndexDetailsPendingAdd, false, false, MetadataUtil.PENDING_ADD_OP, Creator.DEFAULT_CREATOR);
 
@@ -5278,9 +5302,12 @@ protected void doAnalyzeDataset(MetadataProvider metadataProvider, AnalyzeStatem
             }
             DatasetStreamStats stats = new DatasetStreamStats(opStats.get(0));
 
+            LOGGER.info("ANALYZED statement stats: coldReads: {} -- pinnedPages: {} -- cloudPageReads:{}",
+                    stats.getColdReads(), stats.getPinnedPages(), stats.getCloudPageReads());
+
             Index.SampleIndexDetails newIndexDetailsFinal = new Index.SampleIndexDetails(dsDetails.getPrimaryKey(),
                     dsDetails.getKeySourceIndicator(), dsDetails.getPrimaryKeyType(), sampleCardinalityTarget,
-                    stats.getCardinality(), stats.getAvgTupleSize(), sampleSeed, stats.getIndexesStats());
+                    stats.getCardinality(), stats.getAvgTupleSize(), sampleSeed, sampleMethod, stats.getIndexesStats());
             Index newIndexFinal = new Index(databaseName, dataverseName, datasetName, newIndexName, sampleIndexType,
                     newIndexDetailsFinal, false, false, MetadataUtil.PENDING_NO_OP, Creator.DEFAULT_CREATOR);
 

asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudStorageUnstableTest.java

@@ -23,6 +23,7 @@
 import static org.apache.asterix.test.cloud_storage.CloudStorageTest.MOCK_SERVER_HOSTNAME_FRAGMENT;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 import java.util.Random;
@@ -64,6 +65,9 @@ public class CloudStorageUnstableTest {
     private static final String CONFIG_FILE_NAME = "src/test/resources/cc-cloud-storage.conf";
     private static final String DELTA_RESULT_PATH = "results_cloud";
     private static final String EXCLUDED_TESTS = "MP";
+    // Tests excluded from unstable cloud runs because simulated I/O failures (UnstableCloudClient)
+    // cause non-deterministic LSM disk component layouts, making exact sample counts unpredictable.
+    private static final String[] UNSTABLE_DENY_LIST = { "ddl: analyze-dataset-1" };
 
     public CloudStorageUnstableTest(TestCaseContext tcCtx) {
         this.tcCtx = tcCtx;
@@ -108,7 +112,8 @@ public static Collection<Object[]> tests() throws Exception {
     @Test
     public void test() throws Exception {
         List<TestCase.CompilationUnit> cu = tcCtx.getTestCase().getCompilationUnit();
-        Assume.assumeTrue(cu.size() > 1 || !EXCLUDED_TESTS.equals(getText(cu.get(0).getDescription())));
+        Assume.assumeTrue(cu.size() > 1 || (!EXCLUDED_TESTS.equals(getText(cu.get(0).getDescription()))
+                && !Arrays.stream(UNSTABLE_DENY_LIST).anyMatch(s -> tcCtx.toString().contains(s))));
         LangExecutionUtil.test(tcCtx);
     }
 

asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/GCSCloudStorageUnstableTest.java

@@ -23,6 +23,7 @@
 import static org.apache.asterix.api.common.LocalCloudUtil.MOCK_SERVER_REGION;
 import static org.apache.asterix.test.cloud_storage.CloudStorageGCSTest.S3_ONLY;
 
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 
@@ -74,6 +75,9 @@ public class GCSCloudStorageUnstableTest {
     private static final String CONFIG_FILE_NAME = "src/test/resources/cc-cloud-storage-gcs.conf";
     private static final String DELTA_RESULT_PATH = "results_cloud";
     private static final String EXCLUDED_TESTS = "MP";
+    // Tests excluded from unstable cloud runs because simulated I/O failures (UnstableCloudClient)
+    // cause non-deterministic LSM disk component layouts, making exact sample counts unpredictable.
+    private static final String[] UNSTABLE_DENY_LIST = { "ddl: analyze-dataset-1" };
     public static final String MOCK_SERVER_HOSTNAME = "http://127.0.0.1:24443";
     private static final String MOCK_SERVER_PROJECT_ID = "asterixdb-gcs-test-project-id";
 
@@ -111,7 +115,8 @@ public static Collection<Object[]> tests() throws Exception {
     public void test() throws Exception {
         List<TestCase.CompilationUnit> cu = tcCtx.getTestCase().getCompilationUnit();
         Assume.assumeTrue(cu.size() > 1 || (!EXCLUDED_TESTS.equals(getText(cu.get(0).getDescription()))
-                && !S3_ONLY.equals(getText(cu.get(0).getDescription()))));
+                && !S3_ONLY.equals(getText(cu.get(0).getDescription()))
+                && !Arrays.stream(UNSTABLE_DENY_LIST).anyMatch(s -> tcCtx.toString().contains(s))));
         LangExecutionUtil.test(tcCtx);
         for (NodeControllerService nc : ExecutionTestUtil.integrationUtil.ncs) {
             IDatasetLifecycleManager lifecycleManager =

asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/AnalyzingTestExecutor.java

@@ -20,13 +20,18 @@
 
 import static java.nio.charset.StandardCharsets.UTF_8;
 
+import java.io.BufferedReader;
+import java.io.File;
 import java.io.InputStream;
+import java.util.List;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 
 import org.apache.asterix.testframework.context.TestCaseContext;
 import org.apache.asterix.testframework.xml.TestCase;
 import org.apache.commons.io.IOUtils;
+import org.apache.hyracks.util.annotations.AiProvenance;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
@@ -76,6 +81,19 @@ public ExtractedResult executeSqlppUpdateOrDdl(String statement, TestCaseContext
         return res;
     }
 
+    @Override
+    @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_4_8, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.ASSISTED, notes = "Compare plans without cost: ANALYZE uses random LSM sampling, so CBO cardinality/cost estimates are not reproducible run-to-run")
+    public void runScriptAndCompareWithResultPlan(File scriptFile, BufferedReader readerExpected,
+            BufferedReader readerActual) throws Exception {
+        // This suite runs ANALYZE (random leaf sampling) before planning, so the CBO
+        // cardinality/op-cost/total-cost estimates vary run-to-run (the LSM component layout
+        // differs, so the seeded sample draws different tuples). Compare plan structure only,
+        // stripping the cost annotations, which are not reproducible under random sampling.
+        List<String> expectedLines = readerExpected.lines().collect(Collectors.toList());
+        List<String> actualLines = readerActual.lines().collect(Collectors.toList());
+        TestHelper.comparePlansWithoutCost(expectedLines, actualLines, scriptFile);
+    }
+
     private void analyzeFromRegex(Matcher m, String dv, int pos) throws Exception {
         while (m.find()) {
             String ds = m.group(pos);

asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/SqlppAnalyzedExecutionTest.java

@@ -46,7 +46,11 @@ public class SqlppAnalyzedExecutionTest {
     private final String[] denyList = { "synonym: synonym-01", "ddl: analyze-dataset-1", "misc: dump_index",
             "array-index: composite-index-queries", "filters: upsert", "column: analyze-dataset",
             "column: filter/boolean", "column: filter/sql-compat", "ddl: analyze-dataset-with-indexes",
-            "warnings: cardinality-hint-warning", "comparison: incomparable_types" };
+            "warnings: cardinality-hint-warning", "comparison: incomparable_types",
+            // MB-72758: under random-sample-based stats the index-intersection cost decision is not
+            // reproducible, so this plan's structure (INTERSECT vs single-index) varies run-to-run.
+            // Excluded from the analyzed suite; still covered as a result test in the regular suite.
+            "index-selection: secondary-index-intersection-01" };
 
     @BeforeClass
     public static void setUp() throws Exception {

asterixdb/asterix-app/src/test/resources/cc-analyze.conf

@@ -54,6 +54,8 @@ credential.file=src/test/resources/security/passwd
 log.dir = logs/
 log.level = INFO
 compiler.groupmemory=64MB
+# ANALYZE statement needs more memory for sorting
+compiler.sortmemory=128MB
 storage.buffercache.pagesize=32KB
 compiler.internal.sanitycheck=true
 compiler.ordered.fields=false