fix: DH-20625: SelectAndViewAnalyzer track barriers for real and aliased columns (#7328)

This PR fixes issue
[DH-20625](https://deephaven.atlassian.net/browse/DH-20625) by ensuring
that the SelectAndViewAnalyzer properly tracks barriers for both real
and aliased columns. Previously, PreserveColumnLayer instances were
added directly to the context without registering their declared
barriers in the barrierToLayerIndex map, which could lead to incorrect
barrier tracking.

[DH-20625]:
https://deephaven.atlassian.net/browse/DH-20625?atlOrigin=eyJpIjoiNWRkNTljNzYxNjVmNDY3MDlhMDU5Y2ZhYzA5YTRkZjUiLCJwIjoiZ2l0aHViLWNvbS1KU1cifQ

---------

Co-authored-by: Jianfeng Mao <[email protected]>
Co-authored-by: margaretkennedy <[email protected]>

Author: 3 people

engine/table/src/main/java/io/deephaven/engine/table/impl/select/analyzers/SelectAndViewAnalyzer.java

@@ -240,13 +240,15 @@ public static AnalyzerContext createContext(
             final SourceColumn realColumn = sc.maybeGetSourceColumn().orElse(null);
             if (realColumn != null) {
                 if (shouldPreserve(sc.getDataView())) {
-                    context.addLayer(new PreserveColumnLayer(context, sc, sc.getDataView(), distinctDeps, mcsBuilder));
+                    addDeclaredBarriersToMap(sc, barrierToLayerIndex,
+                            new PreserveColumnLayer(context, sc, sc.getDataView(), distinctDeps, mcsBuilder), context);
                     continue;
                 }
                 // look for an existing alias that can be preserved instead
                 final ColumnSource<?> alias = resultAlias.get(realColumn.getSourceName());
                 if (alias != null) {
-                    context.addLayer(new PreserveColumnLayer(context, sc, alias, distinctDeps, mcsBuilder));
+                    addDeclaredBarriersToMap(sc, barrierToLayerIndex,
+                            new PreserveColumnLayer(context, sc, alias, distinctDeps, mcsBuilder), context);
                     continue;
                 }
             }

engine/table/src/test/java/io/deephaven/engine/table/impl/QueryTableSelectBarrierTest.java

@@ -648,6 +648,74 @@ public void testMergingSelectable() {
         assertEquals(ss2.respectedBarriers().length, ss2.withRespectedBarriers(c, c).respectedBarriers().length);
     }
 
+    @Test
+    public void testDH20625() {
+        final Table source = TableTools.emptyTable(1_000)
+                .update("a = ii % 5", "b = ii % 11", "c = ii % 17", "d = ii");
+
+        final Object b1 = new Object();
+        final Object b2 = new Object();
+
+        Table result;
+
+        // update with a, c passed as real columns
+        result = source.update(List.of(
+                Selectable.parse("a").withDeclaredBarriers(b1, b2),
+                Selectable.parse("c").withRespectedBarriers(b2).withSerial(),
+                Selectable.parse("Sum = a + b + c").withRespectedBarriers(b1)));
+        // simple test, verify we have the same number of rows
+        assertEquals(source.size(), result.size());
+
+        // select with a, c passed as real columns
+        result = source.select(List.of(
+                Selectable.parse("a").withDeclaredBarriers(b1, b2),
+                Selectable.parse("c").withRespectedBarriers(b2).withSerial(),
+                Selectable.parse("Sum = a + b + c").withRespectedBarriers(b1)));
+        // simple test, verify we have the same number of rows
+        assertEquals(source.size(), result.size());
+    }
+
+    @Test
+    public void testDH20625_proxy() {
+        // Create a partitioned table with 4 partitions
+        final Table source = TableTools.emptyTable(1_000)
+                .update("a = ii % 5", "b = ii % 11", "c = ii % 17", "d = ii");
+        final PartitionedTable pt = source.partitionBy("a");
+
+        final Object b1 = new Object();
+        final Object b2 = new Object();
+
+        PartitionedTable.Proxy result_proxy;
+        // update test
+        result_proxy = pt.proxy().update(List.of(
+                Selectable.parse("a").withDeclaredBarriers(b1, b2),
+                Selectable.parse("c").withRespectedBarriers(b2).withSerial(),
+                Selectable.parse("Sum = a + b + c").withRespectedBarriers(b1)));
+        assertEquals(pt.constituents().length, result_proxy.target().constituents().length);
+        for (int i = 0; i < pt.constituents().length; i++) {
+            final Table ct = pt.constituents()[i];
+            final Table rct = result_proxy.target().constituents()[i];
+
+            // simple test, verify we have the same number of rows
+            assertEquals(ct.size(), rct.size());
+        }
+
+        // select test
+        result_proxy = pt.proxy().select(List.of(
+                Selectable.parse("a").withDeclaredBarriers(b1, b2),
+                Selectable.parse("c").withRespectedBarriers(b2).withSerial(),
+                Selectable.parse("Sum = a + b + c").withRespectedBarriers(b1)));
+        assertEquals(pt.constituents().length, result_proxy.target().constituents().length);
+        for (int i = 0; i < pt.constituents().length; i++) {
+            final Table ct = pt.constituents()[i];
+            final Table rct = result_proxy.target().constituents()[i];
+
+            // simple test, verify we have the same number of rows
+            assertEquals(ct.size(), rct.size());
+        }
+    }
+
+
     private static void checkIndividualSums(int size, Table u) {
         final long expectedSumA = (((long) size - 1) * size) / 2;
         final Table us = u.aggBy(AggSum("A", "B"));

py/server/tests/test_pt_proxy.py

@@ -365,7 +365,6 @@ def my_fn(vals):
             ptp = agg_by_formula()
             self.assertIsNotNone(ptp)
 
-    @unittest.skip("https://deephaven.atlassian.net/browse/DH-20625")
     def test_update_select_concurrency_control(self):
         ops = [
             PartitionedTableProxy.update,