Merge branch 'master' into fix/mask-sensitive-data-in-debug-log-for-failed-import-validation

Signed-off-by: hainenber <dotronghai96@gmail.com>

.pre-commit-config.yaml

@@ -63,7 +63,7 @@ repos:
     hooks:
       - id: prettier-frontend
         name: prettier (frontend)
-        entry: bash -c 'cd superset-frontend && for file in "$@"; do npx prettier --write "${file#superset-frontend/}"; done'
+        entry: bash -c 'cd superset-frontend && files=(); for f in "$@"; do files+=("${f#superset-frontend/}"); done; npx prettier --write -- "${files[@]}"' --
         language: system
         pass_filenames: true
         files: ^superset-frontend/.*\.(js|jsx|ts|tsx|css|scss|sass|json)$

pyproject.toml

@@ -161,11 +161,11 @@ fastmcp = [
     # heuristic that under-counts JSON-heavy MCP responses.
     "tiktoken>=0.13.0,<1.0",
 ]
-firebird = ["sqlalchemy-firebird>=0.7.0, <2.2"]
+firebird = ["sqlalchemy-firebird>=0.8.0, <2.2"]
 firebolt = ["firebolt-sqlalchemy>=1.0.0, <2"]
 gevent = ["gevent>=26.4.0"]
 gsheets = ["shillelagh[gsheetsapi]>=1.4.4, <2"]
-hana = ["hdbcli==2.28.21", "sqlalchemy_hana==0.4.0"]
+hana = ["hdbcli==2.28.21", "sqlalchemy_hana==3.0.3"]
 hive = [
     "pyhive[hive]>=0.6.5;python_version<'3.11'",
     "pyhive[hive_pure_sasl]>=0.7.0",

superset-frontend/plugins/plugin-chart-pivot-table/src/PivotTableChart.tsx

@@ -28,6 +28,7 @@ import {
   BinaryQueryObjectFilterClause,
   Currency,
   CurrencyFormatter,
+  DataRecord,
   DataRecordValue,
   FeatureFlag,
   getColumnLabel,
@@ -43,6 +44,7 @@ import {
 import { styled, useTheme } from '@apache-superset/core/theme';
 import { aggregatorTemplates, PivotTable, sortAs } from './react-pivottable';
 import {
+  DateFormatter,
   FilterType,
   MetricsLayoutEnum,
   PivotTableProps,
@@ -218,6 +220,46 @@ const aggregatorsFactory = (formatter: NumberFormatter) => ({
   ),
 });
 
+const getDrillFilterValue = (
+  value: string,
+  formatter: DateFormatter | undefined,
+): string | number => {
+  if (formatter && value.trim() !== '' && Number.isFinite(Number(value))) {
+    return Number(value);
+  }
+  return value;
+};
+
+const getCrossFilterValue = (
+  value: DataRecordValue,
+  formatter: DateFormatter | undefined,
+): DataRecordValue => {
+  if (
+    formatter &&
+    typeof value === 'string' &&
+    value.trim() !== '' &&
+    Number.isFinite(Number(value))
+  ) {
+    return Number(value);
+  }
+  return value;
+};
+
+const getDrillFilterFormattedValue = (
+  value: string,
+  formatter: DateFormatter | undefined,
+): string => {
+  const valueToFormat: DataRecordValue =
+    value.trim() !== '' && Number.isFinite(Number(value))
+      ? Number(value)
+      : value;
+  return (
+    (formatter as ((value: DataRecordValue) => string) | undefined)?.(
+      valueToFormat,
+    ) || String(value)
+  );
+};
+
 /* If you change this logic, please update the corresponding Python
  * function (https://github.com/apache/superset/blob/master/superset/charts/post_processing.py),
  * or reach out to @betodealmeida.
@@ -342,7 +384,7 @@ export default function PivotTableChart(props: PivotTableProps) {
   const unpivotedData = useMemo(
     () =>
       data.reduce(
-        (acc: Record<string, any>[], record: Record<string, any>) => [
+        (acc: DataRecord[], record: DataRecord) => [
           ...acc,
           ...metricNames
             .map((name: string) => ({
@@ -419,10 +461,16 @@ export default function PivotTableChart(props: PivotTableProps) {
                       col,
                       op: 'IS NULL',
                     };
+                  // Resolve the formatter by the header key/label so adhoc
+                  // temporal groupby columns (where `col` is an object, not a
+                  // string) still get epoch coercion, matching physical columns.
+                  const formatter = dateFormatters[key];
                   return {
                     col,
                     op: 'IN',
-                    val: val as (string | number | boolean)[],
+                    val: (val as DataRecordValue[]).map(value =>
+                      getCrossFilterValue(value, formatter),
+                    ) as (string | number | boolean)[],
                   };
                 }),
         },
@@ -436,7 +484,7 @@ export default function PivotTableChart(props: PivotTableProps) {
         },
       });
     },
-    [groupbyColumnsRaw, groupbyRowsRaw, setDataMask],
+    [dateFormatters, groupbyColumnsRaw, groupbyRowsRaw, setDataMask],
   );
 
   const isActiveFilterValue = useCallback(
@@ -492,10 +540,17 @@ export default function PivotTableChart(props: PivotTableProps) {
                         col,
                         op: 'IS NULL' as const,
                       };
+                    // Resolve the formatter by the header key/label so adhoc
+                    // temporal groupby columns (where `col` is an object, not a
+                    // string) still get epoch coercion, matching physical
+                    // columns.
+                    const formatter = dateFormatters[key];
                     return {
                       col,
                       op: 'IN' as const,
-                      val: val as (string | number | boolean)[],
+                      val: (val as DataRecordValue[]).map(value =>
+                        getCrossFilterValue(value, formatter),
+                      ) as (string | number | boolean)[],
                     };
                   }),
           },
@@ -511,7 +566,13 @@ export default function PivotTableChart(props: PivotTableProps) {
         isCurrentValueSelected: isActiveFilterValue(key, val),
       };
     },
-    [groupbyColumnsRaw, groupbyRowsRaw, isActiveFilterValue, selectedFilters],
+    [
+      dateFormatters,
+      groupbyColumnsRaw,
+      groupbyRowsRaw,
+      isActiveFilterValue,
+      selectedFilters,
+    ],
   );
 
   const toggleFilter = useCallback(
@@ -535,7 +596,10 @@ export default function PivotTableChart(props: PivotTableProps) {
       const filtersCopy = { ...filters };
       delete filtersCopy[METRIC_KEY];
 
-      const filtersEntries = Object.entries(filtersCopy);
+      const filtersEntries = Object.entries(filtersCopy) as [
+        string,
+        DataRecordValue,
+      ][];
       if (filtersEntries.length === 0) {
         return;
       }
@@ -637,12 +701,12 @@ export default function PivotTableChart(props: PivotTableProps) {
           colKey.forEach((val, i) => {
             const col = cols[i];
             const formatter = dateFormatters[col];
-            const formattedVal = formatter?.(Number(val)) || String(val);
+            const formattedVal = getDrillFilterFormattedValue(val, formatter);
             if (i > 0) {
               drillToDetailFilters.push({
                 col,
                 op: '==',
-                val,
+                val: getDrillFilterValue(val, formatter),
                 formattedVal,
                 grain: formatter ? timeGrainSqla : undefined,
               });
@@ -653,11 +717,11 @@ export default function PivotTableChart(props: PivotTableProps) {
           rowKey.forEach((val, i) => {
             const col = rows[i];
             const formatter = dateFormatters[col];
-            const formattedVal = formatter?.(Number(val)) || String(val);
+            const formattedVal = getDrillFilterFormattedValue(val, formatter);
             drillToDetailFilters.push({
               col,
               op: '==',
-              val,
+              val: getDrillFilterValue(val, formatter),
               formattedVal,
               grain: formatter ? timeGrainSqla : undefined,
             });

superset-frontend/plugins/plugin-chart-pivot-table/test/PivotTableChart.test.tsx

@@ -0,0 +1,370 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import type { ReactElement } from 'react';
+import '@testing-library/jest-dom';
+import { fireEvent, render, screen } from '@testing-library/react';
+import { supersetTheme, ThemeProvider } from '@apache-superset/core/theme';
+import {
+  TimeGranularity,
+  type DataRecordValue,
+  type QueryFormColumn,
+} from '@superset-ui/core';
+import PivotTableChart from '../src/PivotTableChart';
+import { MetricsLayoutEnum, type PivotTableProps } from '../src/types';
+
+function renderWithTheme(ui: ReactElement) {
+  return render(<ThemeProvider theme={supersetTheme}>{ui}</ThemeProvider>);
+}
+
+function createProps(
+  overrides: Partial<PivotTableProps> = {},
+): PivotTableProps {
+  return {
+    data: [],
+    height: 400,
+    width: 600,
+    margin: 0,
+    groupbyRows: [],
+    groupbyColumns: [],
+    metrics: ['value'],
+    tableRenderer: 'Table',
+    colOrder: 'key_a_to_z',
+    rowOrder: 'key_a_to_z',
+    aggregateFunction: 'Count',
+    transposePivot: false,
+    combineMetric: false,
+    rowSubtotalPosition: false,
+    colSubtotalPosition: false,
+    colTotals: false,
+    colSubTotals: false,
+    rowTotals: false,
+    rowSubTotals: false,
+    valueFormat: 'SMART_NUMBER',
+    currencyFormat: { symbol: 'USD', symbolPosition: 'prefix' },
+    setDataMask: jest.fn(),
+    emitCrossFilters: true,
+    selectedFilters: {},
+    verboseMap: {},
+    columnFormats: {},
+    currencyFormats: {},
+    metricsLayout: MetricsLayoutEnum.COLUMNS,
+    metricColorFormatters: [],
+    dateFormatters: {},
+    legacy_order_by: null,
+    order_desc: false,
+    timeGrainSqla: TimeGranularity.DAY,
+    allowRenderHtml: false,
+    ...overrides,
+  };
+}
+
+test('emits numeric temporal values for drill-to-detail filters on formatted row headers', () => {
+  const onContextMenu = jest.fn();
+  const timestamp = 1778630400000;
+  const props: PivotTableProps = {
+    data: [{ install_date: String(timestamp), value: 1 }],
+    height: 400,
+    width: 600,
+    margin: 0,
+    groupbyRows: ['install_date'],
+    groupbyColumns: [],
+    metrics: ['value'],
+    tableRenderer: 'Table',
+    colOrder: 'key_a_to_z',
+    rowOrder: 'key_a_to_z',
+    aggregateFunction: 'Count',
+    transposePivot: false,
+    combineMetric: false,
+    rowSubtotalPosition: false,
+    colSubtotalPosition: false,
+    colTotals: false,
+    colSubTotals: false,
+    rowTotals: false,
+    rowSubTotals: false,
+    valueFormat: 'SMART_NUMBER',
+    currencyFormat: { symbol: 'USD', symbolPosition: 'prefix' },
+    setDataMask: jest.fn(),
+    emitCrossFilters: true,
+    selectedFilters: {},
+    verboseMap: {},
+    columnFormats: {},
+    currencyFormats: {},
+    metricsLayout: MetricsLayoutEnum.COLUMNS,
+    metricColorFormatters: [],
+    dateFormatters: {
+      install_date: (value: DataRecordValue) =>
+        new Date(Number(value)).toISOString().slice(0, 10),
+    },
+    legacy_order_by: null,
+    order_desc: false,
+    onContextMenu,
+    timeGrainSqla: TimeGranularity.DAY,
+    allowRenderHtml: false,
+  };
+
+  renderWithTheme(<PivotTableChart {...props} />);
+
+  const rowHeader = screen.getByText('2026-05-13').closest('th');
+  expect(rowHeader).not.toBeNull();
+  fireEvent.contextMenu(rowHeader!);
+
+  expect(onContextMenu).toHaveBeenCalledTimes(1);
+  const contextMenuFilters = onContextMenu.mock.calls[0][2];
+  expect(contextMenuFilters?.drillToDetail).toEqual([
+    {
+      col: 'install_date',
+      op: '==',
+      val: timestamp,
+      formattedVal: '2026-05-13',
+      grain: TimeGranularity.DAY,
+    },
+  ]);
+});
+
+test('keeps non-numeric temporal values for drill-to-detail formatted labels', () => {
+  const onContextMenu = jest.fn();
+  const dateValue = '2024-01-01';
+  const props: PivotTableProps = {
+    data: [{ install_date: dateValue, value: 1 }],
+    height: 400,
+    width: 600,
+    margin: 0,
+    groupbyRows: ['install_date'],
+    groupbyColumns: [],
+    metrics: ['value'],
+    tableRenderer: 'Table',
+    colOrder: 'key_a_to_z',
+    rowOrder: 'key_a_to_z',
+    aggregateFunction: 'Count',
+    transposePivot: false,
+    combineMetric: false,
+    rowSubtotalPosition: false,
+    colSubtotalPosition: false,
+    colTotals: false,
+    colSubTotals: false,
+    rowTotals: false,
+    rowSubTotals: false,
+    valueFormat: 'SMART_NUMBER',
+    currencyFormat: { symbol: 'USD', symbolPosition: 'prefix' },
+    setDataMask: jest.fn(),
+    emitCrossFilters: true,
+    selectedFilters: {},
+    verboseMap: {},
+    columnFormats: {},
+    currencyFormats: {},
+    metricsLayout: MetricsLayoutEnum.COLUMNS,
+    metricColorFormatters: [],
+    dateFormatters: {
+      install_date: (value: DataRecordValue) => String(value),
+    },
+    legacy_order_by: null,
+    order_desc: false,
+    onContextMenu,
+    timeGrainSqla: TimeGranularity.DAY,
+    allowRenderHtml: false,
+  };
+
+  renderWithTheme(<PivotTableChart {...props} />);
+
+  const rowHeader = screen.getByText(dateValue).closest('th');
+  expect(rowHeader).not.toBeNull();
+  fireEvent.contextMenu(rowHeader!);
+
+  expect(onContextMenu).toHaveBeenCalledTimes(1);
+  const contextMenuFilters = onContextMenu.mock.calls[0][2];
+  expect(contextMenuFilters?.drillToDetail).toEqual([
+    {
+      col: 'install_date',
+      op: '==',
+      val: dateValue,
+      formattedVal: dateValue,
+      grain: TimeGranularity.DAY,
+    },
+  ]);
+});
+
+test('keeps non-formatted drill-to-detail values as strings', () => {
+  const onContextMenu = jest.fn();
+  const props = createProps({
+    data: [{ country: 'US', value: 1 }],
+    groupbyRows: ['country'],
+    onContextMenu,
+  });
+
+  renderWithTheme(<PivotTableChart {...props} />);
+
+  const rowHeader = screen.getByText('US').closest('th');
+  expect(rowHeader).not.toBeNull();
+  fireEvent.contextMenu(rowHeader!);
+
+  expect(onContextMenu).toHaveBeenCalledTimes(1);
+  const contextMenuFilters = onContextMenu.mock.calls[0][2];
+  expect(contextMenuFilters?.drillToDetail).toEqual([
+    {
+      col: 'country',
+      op: '==',
+      val: 'US',
+      formattedVal: 'US',
+      grain: undefined,
+    },
+  ]);
+});
+
+test('emits numeric temporal values for cross-filters on formatted row headers', () => {
+  const setDataMask = jest.fn();
+  const timestamp = 1777248000000;
+  const props: PivotTableProps = {
+    data: [{ install_date: String(timestamp), value: 1 }],
+    height: 400,
+    width: 600,
+    margin: 0,
+    groupbyRows: ['install_date'],
+    groupbyColumns: [],
+    metrics: ['value'],
+    tableRenderer: 'Table',
+    colOrder: 'key_a_to_z',
+    rowOrder: 'key_a_to_z',
+    aggregateFunction: 'Count',
+    transposePivot: false,
+    combineMetric: false,
+    rowSubtotalPosition: false,
+    colSubtotalPosition: false,
+    colTotals: false,
+    colSubTotals: false,
+    rowTotals: false,
+    rowSubTotals: false,
+    valueFormat: 'SMART_NUMBER',
+    currencyFormat: { symbol: 'USD', symbolPosition: 'prefix' },
+    setDataMask,
+    emitCrossFilters: true,
+    selectedFilters: {},
+    verboseMap: {},
+    columnFormats: {},
+    currencyFormats: {},
+    metricsLayout: MetricsLayoutEnum.COLUMNS,
+    metricColorFormatters: [],
+    dateFormatters: {
+      install_date: (value: DataRecordValue) =>
+        new Date(Number(value)).toISOString().slice(0, 10),
+    },
+    legacy_order_by: null,
+    order_desc: false,
+    timeGrainSqla: TimeGranularity.DAY,
+    allowRenderHtml: false,
+  };
+
+  renderWithTheme(<PivotTableChart {...props} />);
+
+  const rowHeader = screen.getByText('2026-04-27').closest('th');
+  expect(rowHeader).not.toBeNull();
+  fireEvent.click(rowHeader!);
+
+  expect(setDataMask).toHaveBeenCalledWith(
+    expect.objectContaining({
+      extraFormData: {
+        filters: [
+          {
+            col: 'install_date',
+            op: 'IN',
+            val: [timestamp],
+          },
+        ],
+      },
+    }),
+  );
+});
+
+test('keeps non-numeric temporal values for cross-filters on formatted row headers', () => {
+  const setDataMask = jest.fn();
+  const dateValue = '2024-01-01';
+  const props = createProps({
+    data: [{ install_date: dateValue, value: 1 }],
+    groupbyRows: ['install_date'],
+    setDataMask,
+    dateFormatters: {
+      install_date: (value: DataRecordValue) => String(value),
+    },
+  });
+
+  renderWithTheme(<PivotTableChart {...props} />);
+
+  const rowHeader = screen.getByText(dateValue).closest('th');
+  expect(rowHeader).not.toBeNull();
+  fireEvent.click(rowHeader!);
+
+  expect(setDataMask).toHaveBeenCalledWith(
+    expect.objectContaining({
+      extraFormData: {
+        filters: [
+          {
+            col: 'install_date',
+            op: 'IN',
+            val: [dateValue],
+          },
+        ],
+      },
+    }),
+  );
+});
+
+test('emits drill filters from formatted column headers', () => {
+  const onContextMenu = jest.fn();
+  const timestamp = 1778630400000;
+  const adhocColumn: QueryFormColumn = {
+    label: 'Install date expression',
+    sqlExpression: 'install_date',
+    expressionType: 'SQL',
+  };
+  const props = createProps({
+    data: [{ 'Install date expression': String(timestamp), value: 1 }],
+    groupbyColumns: [adhocColumn],
+    dateFormatters: {
+      'Install date expression': (value: DataRecordValue) =>
+        new Date(Number(value)).toISOString().slice(0, 10),
+    },
+    onContextMenu,
+  });
+
+  renderWithTheme(<PivotTableChart {...props} />);
+
+  const columnHeader = screen.getByText('2026-05-13').closest('th');
+  expect(columnHeader).not.toBeNull();
+  fireEvent.contextMenu(columnHeader!);
+
+  expect(onContextMenu).toHaveBeenCalledTimes(1);
+  const contextMenuFilters = onContextMenu.mock.calls[0][2];
+  expect(contextMenuFilters?.drillToDetail).toEqual([
+    {
+      col: 'Install date expression',
+      op: '==',
+      val: timestamp,
+      formattedVal: '2026-05-13',
+      grain: TimeGranularity.DAY,
+    },
+  ]);
+  expect(contextMenuFilters?.crossFilter?.dataMask.extraFormData).toEqual({
+    filters: [
+      {
+        col: adhocColumn,
+        op: 'IN',
+        val: [timestamp],
+      },
+    ],
+  });
+});

superset-frontend/plugins/plugin-chart-table/src/TableChart.tsx

@@ -108,6 +108,25 @@ interface TableSize {
   height: number;
 }
 
+const getCrossFilterValue = (
+  value: DataRecordValue,
+  column: DataColumnMeta | undefined,
+): DataRecordValue => {
+  const input = value instanceof DateWithFormatter ? value.input : value;
+  if (
+    column?.dataType === GenericDataType.Temporal &&
+    typeof input === 'string' &&
+    input.trim() !== '' &&
+    Number.isFinite(Number(input))
+  ) {
+    return Number(input);
+  }
+  if (value instanceof Date) {
+    return value.getTime();
+  }
+  return value;
+};
+
 const ACTION_KEYS = {
   enter: 'Enter',
   spacebar: 'Spacebar',
@@ -533,6 +552,9 @@ export default function TableChart<D extends DataRecord = DataRecord>(
                     // so that cross-filters work on the receiving chart
                     const resolvedCol = columnLabelToNameMap[col] ?? col;
                     const val = ensureIsArray(updatedFilters?.[col]);
+                    const column = columnsMeta.find(
+                      columnMeta => columnMeta.key === col,
+                    );
                     if (
                       !val.length ||
                       val[0] === null ||
@@ -546,9 +568,7 @@ export default function TableChart<D extends DataRecord = DataRecord>(
                     return {
                       col: resolvedCol,
                       op: 'IN' as const,
-                      val: val.map(el =>
-                        el instanceof Date ? el.getTime() : el!,
-                      ),
+                      val: val.map(el => getCrossFilterValue(el!, column)),
                       grain: resolvedCol === DTTM_ALIAS ? timeGrain : undefined,
                     };
                   }),
@@ -571,6 +591,7 @@ export default function TableChart<D extends DataRecord = DataRecord>(
       timestampFormatter,
       timeGrain,
       columnLabelToNameMap,
+      columnsMeta,
     ],
   );
 
@@ -1317,7 +1338,6 @@ export default function TableChart<D extends DataRecord = DataRecord>(
                 col.toggleSortBy();
               }
             }}
-            role="columnheader button"
             onClick={onClick}
             data-column-name={col.id}
             {...(allowRearrangeColumns && {

superset-frontend/plugins/plugin-chart-table/test/TableChart.test.tsx

@@ -31,6 +31,7 @@ import {
 } from '@superset-ui/core/spec';
 import { cloneDeep } from 'lodash-es';
 import {
+  type DataMask,
   QueryMode,
   TimeGranularity,
   SMART_DATE_ID,
@@ -1843,6 +1844,54 @@ describe('plugin-chart-table', () => {
         expect(secondCallArg.extraFormData.filters).toEqual([]);
       });
 
+      test('clicking a temporal numeric-string cell emits numeric cross-filter values', () => {
+        const setDataMask = jest.fn<void, [DataMask]>();
+        const timestamp = 1777248000000;
+        const props = transformProps({
+          ...testData.basic,
+          hooks: { setDataMask },
+          emitCrossFilters: true,
+        });
+
+        render(
+          <ProviderWrapper>
+            <TableChart
+              {...props}
+              data={[{ install_date: String(timestamp) }]}
+              columns={[
+                {
+                  key: 'install_date',
+                  label: 'install_date',
+                  dataType: GenericDataType.Temporal,
+                  isNumeric: false,
+                  isMetric: false,
+                  isPercentMetric: false,
+                  formatter: String,
+                  config: {},
+                },
+              ]}
+              emitCrossFilters
+              setDataMask={setDataMask}
+              sticky={false}
+            />
+          </ProviderWrapper>,
+        );
+
+        fireEvent.click(screen.getByText(String(timestamp)));
+
+        const crossFilterCall = setDataMask.mock.calls.find(
+          call => call[0].filterState?.filters,
+        );
+        expect(crossFilterCall).toBeDefined();
+        expect(crossFilterCall?.[0].extraFormData?.filters).toEqual([
+          {
+            col: 'install_date',
+            op: 'IN',
+            val: [timestamp],
+          },
+        ]);
+      });
+
       test('cross-filter toggle works with DateWithFormatter values', () => {
         const setDataMask = jest.fn();
         const props = transformProps({

superset-frontend/src/explore/components/controls/TextControl/index.tsx

@@ -27,6 +27,7 @@ type InputValueType = string | number;
 export interface TextControlProps<T extends InputValueType = InputValueType> {
   name?: string;
   label?: string;
+  ariaLabel?: string;
   description?: string;
   disabled?: boolean;
   isFloat?: boolean;
@@ -48,6 +49,7 @@ const safeStringify = (value?: InputValueType | null) =>
 function TextControl<T extends InputValueType = InputValueType>({
   name,
   label,
+  ariaLabel,
   description,
   disabled,
   isFloat,
@@ -143,7 +145,7 @@ function TextControl<T extends InputValueType = InputValueType>({
         onFocus={onFocus}
         value={displayValue}
         disabled={disabled}
-        aria-label={label}
+        aria-label={ariaLabel ?? label}
       />
     </div>
   );

superset-frontend/src/explore/components/controls/ViewportControl.tsx

@@ -78,6 +78,7 @@ export default function ViewportControl({
         value={value?.[ctrl]}
         onChange={(ctrlValue: number) => handleChange(ctrl, ctrlValue)}
         isFloat
+        ariaLabel={ctrl}
       />
     </div>
   );

superset-frontend/src/features/home/RightMenu.tsx

@@ -755,6 +755,7 @@ const RightMenu = ({
             target="_blank"
             rel="noreferrer"
             title={navbarRight.documentation_text || t('Documentation')}
+            aria-label={navbarRight.documentation_text || t('Documentation')}
           >
             {navbarRight.documentation_icon ? (
               <Icons.BookOutlined />
@@ -772,6 +773,7 @@ const RightMenu = ({
             target="_blank"
             rel="noreferrer"
             title={navbarRight.bug_report_text || t('Report a bug')}
+            aria-label={navbarRight.bug_report_text || t('Report a bug')}
           >
             {navbarRight.bug_report_icon ? (
               <i className={navbarRight.bug_report_icon} />

superset/commands/dataset/duplicate.py

@@ -75,43 +75,42 @@ class DuplicateDatasetCommand(CreateMixin, BaseCommand):
                 ),
                 status=404,
             )
-        table = SqlaTable(table_name=table_name, editors=editors)
+        table = SqlaTable()
+        table.override(self._base_model)
+        table.table_name = table_name
+        table.editors = editors
         table.database = database
-        table.schema = self._base_model.schema
-        table.catalog = self._base_model.catalog
-        table.template_params = self._base_model.template_params
-        table.normalize_columns = self._base_model.normalize_columns
-        table.always_filter_main_dttm = self._base_model.always_filter_main_dttm
         table.is_sqllab_view = True
-        table.sql = self._base_model.sql.strip().strip(";")
+        if table.sql:
+            table.sql = table.sql.strip().strip(";")
         db.session.add(table)
-        cols = []
-        for config_ in self._base_model.columns:
-            column_name = config_.column_name
-            col = TableColumn(
-                column_name=column_name,
-                verbose_name=config_.verbose_name,
-                expression=config_.expression,
-                filterable=True,
-                groupby=True,
-                is_dttm=config_.is_dttm,
-                type=config_.type,
-                description=config_.description,
+        table.columns = [
+            TableColumn(
+                column_name=c.column_name,
+                verbose_name=c.verbose_name,
+                expression=c.expression,
+                filterable=c.filterable,
+                groupby=c.groupby,
+                is_dttm=c.is_dttm,
+                type=c.type,
+                description=c.description,
             )
-            cols.append(col)
-        table.columns = cols
-        mets = []
-        for config_ in self._base_model.metrics:
-            metric_name = config_.metric_name
-            met = SqlMetric(
-                metric_name=metric_name,
-                verbose_name=config_.verbose_name,
-                expression=config_.expression,
-                metric_type=config_.metric_type,
-                description=config_.description,
+            for c in self._base_model.columns
+        ]
+        table.metrics = [
+            SqlMetric(
+                metric_name=m.metric_name,
+                verbose_name=m.verbose_name,
+                expression=m.expression,
+                metric_type=m.metric_type,
+                description=m.description,
+                d3format=m.d3format,
+                currency=m.currency,
+                warning_text=m.warning_text,
+                extra=m.extra,
             )
-            mets.append(met)
-        table.metrics = mets
+            for m in self._base_model.metrics
+        ]
         return table
 
     def validate(self) -> None:

superset/commands/importers/v1/utils.py

@@ -149,6 +149,7 @@ def load_configs(
         prefix = file_name.split("/")[0]
         schema = schemas.get(f"{prefix}/")
         if schema:
+            config = None
             try:
                 config = load_yaml(file_name, content)
 
@@ -230,7 +231,21 @@ def load_configs(
                     prefix,
                     exc.messages,
                 )
-                logger.debug("Config content that failed validation: %s", config)
+                # Mask sensitive data before logging out for debug
+                if config:
+                    redacted_config = json.redact_sensitive(
+                        config,
+                        {
+                            "$.password",
+                            "$.ssh_tunnel.password",
+                            "$.ssh_tunnel.private_key",
+                            "$.ssh_tunnel.private_key_password",
+                            "$.masked_encrypted_extra",
+                        },
+                    )
+                    logger.debug(
+                        "Config content that failed validation: %s.", redacted_config
+                    )
                 exc.messages = {file_name: exc.messages}
                 exceptions.append(exc)
 

superset/models/helpers.py

@@ -2710,21 +2710,50 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
         if values is None:
             return None
 
+        temporal_comparison_operators: set[utils.FilterOperator] = {
+            utils.FilterOperator.EQUALS,
+            utils.FilterOperator.NOT_EQUALS,
+            utils.FilterOperator.IN,
+            utils.FilterOperator.NOT_IN,
+            utils.FilterOperator.GREATER_THAN,
+            utils.FilterOperator.LESS_THAN,
+            utils.FilterOperator.GREATER_THAN_OR_EQUALS,
+            utils.FilterOperator.LESS_THAN_OR_EQUALS,
+        }
+
+        def handle_temporal_value(value: FilterValue) -> FilterValue | ColumnElement:
+            if (
+                operator not in temporal_comparison_operators
+                or target_generic_type != utils.GenericDataType.TEMPORAL
+                or target_native_type is None
+                or db_engine_spec is None
+            ):
+                return value
+
+            if isinstance(value, (float, int)) and not isinstance(value, bool):
+                epoch_ms: float = value
+            elif isinstance(value, str) and re.fullmatch(r"[+-]?\d+", value):
+                epoch_ms = int(value)
+            else:
+                return value
+
+            try:
+                dttm = datetime.fromtimestamp(epoch_ms / 1000, tz=timezone.utc).replace(
+                    tzinfo=None
+                )
+            except (OverflowError, OSError, ValueError):
+                return value
+
+            temporal_sql = db_engine_spec.convert_dttm(
+                target_type=target_native_type,
+                dttm=dttm,
+                db_extra=db_extra,
+            )
+            return literal_column(temporal_sql) if temporal_sql is not None else value
+
         def handle_single_value(value: Optional[FilterValue]) -> Optional[FilterValue]:
             if operator == utils.FilterOperator.TEMPORAL_RANGE:
                 return value
-            if (
-                isinstance(value, (float, int))
-                and target_generic_type == utils.GenericDataType.TEMPORAL
-                and target_native_type is not None
-                and db_engine_spec is not None
-            ):
-                value = db_engine_spec.convert_dttm(
-                    target_type=target_native_type,
-                    dttm=datetime.utcfromtimestamp(value / 1000),
-                    db_extra=db_extra,
-                )
-                value = literal_column(value)
             if isinstance(value, str):
                 value = value.strip("\t\n")
 
@@ -2745,6 +2774,9 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
                     return None
                 if value == EMPTY_STRING:
                     return ""
+                value = handle_temporal_value(value)
+            elif value is not None:
+                value = handle_temporal_value(value)
             if target_generic_type == utils.GenericDataType.BOOLEAN:
                 return utils.cast_to_boolean(value)
             return value
@@ -3735,6 +3767,7 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
                     target_native_type=col_type,
                     is_list_target=is_list_target,
                     db_engine_spec=db_engine_spec,
+                    db_extra=self.db_extra,
                 )
 
                 # Get ADVANCED_DATA_TYPES from config when needed

superset/utils/csv.py

@@ -15,7 +15,6 @@
 # specific language governing permissions and limitations
 # under the License.
 import logging
-import re
 import urllib.request
 from typing import Any, Optional, Union
 from urllib.error import URLError
@@ -27,18 +26,37 @@ from superset.utils.core import GenericDataType
 
 logger = logging.getLogger(__name__)
 
-negative_number_re = re.compile(r"^-[0-9.]+$")
+PROBLEMATIC_CSV_PREFIXES: str = "-@+|=%"
 
-# This regex will match if the string starts with:
-#
-#     1. one of -, @, +, |, =, %, a tab, or a carriage return
-#     2. two double quotes immediately followed by one of -, @, +, |, =, %
-#     3. one or more spaces immediately followed by one of -, @, +, |, =, %
-#
-# A leading tab or carriage return is treated as dangerous on its own because
-# some spreadsheet software trims that leading whitespace and then evaluates
-# the remaining cell content as a formula.
-problematic_chars_re = re.compile(r'^(?:"{2}|\s{1,})(?=[\-@+|=%])|^[\-@+|=%\t\r]')
+
+def _starts_with_formula_prefix(value: str) -> bool:
+    first = value[0]
+    if first in PROBLEMATIC_CSV_PREFIXES:
+        return True
+    if first == '"' and len(value) > 2:
+        return value[1] == '"' and value[2] in PROBLEMATIC_CSV_PREFIXES
+    return False
+
+
+def _starts_like_spreadsheet_formula(value: str) -> bool:
+    # A leading tab or carriage return is treated as dangerous on its own
+    # because some spreadsheet software trims that leading whitespace and
+    # then evaluates the remaining cell content as a formula.
+    first = value[0]
+    if first in ("\t", "\r"):
+        return True
+    if first.isspace():
+        stripped = value.lstrip()
+        return bool(stripped) and _starts_with_formula_prefix(stripped)
+    return _starts_with_formula_prefix(value)
+
+
+def _is_negative_number(value: str) -> bool:
+    return (
+        len(value) > 1
+        and value[0] == "-"
+        and all("0" <= character <= "9" or character == "." for character in value[1:])
+    )
 
 
 def escape_value(value: str) -> str:
@@ -47,10 +65,10 @@ def escape_value(value: str) -> str:
 
     http://georgemauer.net/2017/10/07/csv-injection.html
     """
-    needs_escaping = problematic_chars_re.match(value) is not None
-    is_negative_number = negative_number_re.match(value) is not None
+    if not value:
+        return value
 
-    if needs_escaping and not is_negative_number:
+    if _starts_like_spreadsheet_formula(value) and not _is_negative_number(value):
         # Escape pipe to be extra safe as this
         # can lead to remote code execution
         value = value.replace("|", "\\|")

superset/versioning/changes/listener.py

@@ -16,30 +16,11 @@
 # under the License.
 """Session-level listeners that drive ``version_changes`` writes.
 
-Two flush events cooperate, plus two post-commit / post-rollback
-cleanups:
-
-- ``before_flush``: for each versioned entity in ``session.dirty``,
-  reads the pre-save scalar state from the DB via raw SQL inside
-  ``session.no_autoflush`` (same idiom as the baseline listener, not
-  Continuum's internal ``units_of_work`` which is a private API), reads
-  the post-save state from the in-memory ORM object, calls the diff
-  engine, and buffers the resulting :class:`ChangeRecord` list on
-  ``session.info``. This must run before the flush because after the
-  flush the DB already reflects the post-state; we can't recover the
-  pre-state from it.
-
-- ``after_flush``: drains the buffer, resolves the current Continuum
-  transaction id via ``versioning_manager.units_of_work``, and bulk-
-  inserts one ``version_changes`` row per record with a monotonic
-  ``sequence`` number. Records accumulated across multiple before_flush
-  calls within one transaction share the same ``transaction_id`` and
-  contiguous sequence numbers.
-
-- ``after_commit`` / ``after_rollback``: clean up session-scoped
-  state (processed-tx set, ``action_kind`` / ``action_meta`` keys, and
-  the pending-records buffer) so a long-lived session doesn't carry any
-  of it into the next transaction.
+The listeners retain each entity's first pre-flush state, force the final
+flush from ``before_commit``, and write one net initial-to-final semantic
+projection for the Continuum transaction. Completion of the outer transaction
+clears all session-scoped state so long-lived sessions cannot leak versioning
+intent.
 
 Scope:
   - Slice, Dashboard, SqlaTable **scalar fields** (via the cached
@@ -51,8 +32,8 @@ Scope:
 Child-collection diffs (dataset ``TableColumn`` / ``SqlMetric``,
 dashboard ``dashboard_slices``) read the pre- and post-state from
 Continuum shadow tables via the helpers in
-:mod:`superset.versioning.changes.shadow_queries`, executed in
-``after_flush`` once Continuum has written its tx-N rows.
+:mod:`superset.versioning.changes.shadow_queries`, executed during commit
+finalization after the explicit final flush has written Continuum's tx-N rows.
 
 ``session.new`` entities are not processed in this listener:
 operation_type=0 transactions (baseline capture and first-save INSERTs)
@@ -67,7 +48,7 @@ from typing import Any
 import sqlalchemy as sa
 from sqlalchemy import event
 from sqlalchemy.exc import OperationalError, ProgrammingError
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import Session, SessionTransaction
 
 from superset.versioning.changes.shadow_queries import (
     _dashboard_child_records_for_tx_from_shadows,
@@ -75,7 +56,8 @@ from superset.versioning.changes.shadow_queries import (
 )
 from superset.versioning.changes.state import (
     bulk_insert_records,
-    compute_records_for_entity,
+    capture_initial_state,
+    compute_records_from_state,
 )
 from superset.versioning.changes.table import ENTITY_KIND_BY_CLASS_NAME
 from superset.versioning.diff import (
@@ -86,20 +68,9 @@ from superset.versioning.diff import (
 logger = logging.getLogger(__name__)
 
 
-# Key under which the pending-records buffer is stored on ``session.info``.
-# Using ``session.info`` (SQLAlchemy's user-data dict) avoids the need
-# for a module-level WeakKeyDictionary and keeps buffers naturally scoped
-# to the session's lifetime.
-_BUFFER_KEY = "_version_changes_pending"
-
-# Key for the set of Continuum transaction ids whose change records
-# have already been written in this session. ``after_flush`` can fire
-# more than once for a single transaction (e.g. autoflush triggered by
-# a mid-commit query), and our child-diff path reads snapshot tables
-# that don't care about the buffer state — without this marker we'd
-# re-insert the same child records on the second flush and hit the
-# UNIQUE(transaction_id, entity_kind, entity_id, sequence) constraint.
-_PROCESSED_TXS_KEY = "_version_changes_processed_txs"
+# Keys for transaction-scoped state stored on ``session.info``.
+_INITIAL_STATES_KEY = "_version_changes_initial_states"
+_FINALIZING_KEY = "_version_changes_finalizing"
 
 # Key on ``session.info`` that commands set to declare the high-level
 # action that produced the current transaction. Read once per flush by
@@ -112,8 +83,8 @@ _PROCESSED_TXS_KEY = "_version_changes_processed_txs"
 #     db.session.info[ACTION_KIND_KEY] = ACTION_KIND_RESTORE
 #     db.session.commit()
 #
-# The listener pops the key after stamping, and ``after_commit`` /
-# ``after_rollback`` cleanup pop it again as a safety net, so a
+# The listener pops the key after stamping, and outer-transaction cleanup pops
+# it again as a safety net, so a
 # long-lived session can't accidentally carry the value into the next
 # transaction.
 ACTION_KIND_KEY = "_versioning_action_kind"
@@ -199,29 +170,59 @@ def build_action_headline(
 _REGISTERED_SENTINEL = "_versioning_change_listener_registered"
 
 
-def _process_dirty_entity_into_buffer(
+def _capture_dirty_entity_initial_state(
     session: Session,
     obj: Any,
-    buffer: dict[tuple[str, int], list[ChangeRecord]],
+    initial_states: dict[tuple[str, int], tuple[Any, dict[str, Any]]],
 ) -> None:
-    """Compute scalar change records for one dirty entity + append to buffer."""
+    """Retain one dirty entity's first database state for this transaction."""
     entity_kind = ENTITY_KIND_BY_CLASS_NAME.get(type(obj).__name__)
     if entity_kind is None:
         return
     entity_id = getattr(obj, "id", None)
     if entity_id is None:
         return
-    try:
-        records = compute_records_for_entity(session, obj)
-    except Exception:  # pylint: disable=broad-except
-        logger.exception(
-            "version_changes: diff failed for %s id=%s",
-            type(obj).__name__,
-            entity_id,
-        )
+    key = (entity_kind, entity_id)
+    if key in initial_states:
         return
-    if records:
-        buffer.setdefault((entity_kind, entity_id), []).extend(records)
+    if (pre_state := capture_initial_state(session, obj)) is not None:
+        initial_states[key] = (obj, pre_state)
+
+
+def _build_scalar_buffer(
+    initial_states: dict[tuple[str, int], tuple[Any, dict[str, Any]]],
+) -> dict[tuple[str, int], list[ChangeRecord]]:
+    """Build net scalar records from retained initial and final entity states."""
+    buffer: dict[tuple[str, int], list[ChangeRecord]] = {}
+    for key, (obj, pre_state) in initial_states.items():
+        try:
+            records = compute_records_from_state(obj, pre_state)
+        except Exception:  # pylint: disable=broad-except
+            logger.exception(
+                "version_changes: final diff failed for %s id=%s",
+                type(obj).__name__,
+                key[1],
+            )
+            continue
+        if records:
+            buffer[key] = records
+    return buffer
+
+
+def _reset_transaction_state(session: Session) -> None:
+    """Discard versioning intent and retained state after a terminal event."""
+    session.info.pop(ACTION_KIND_KEY, None)
+    session.info.pop(ACTION_META_KEY, None)
+    session.info.pop(_INITIAL_STATES_KEY, None)
+    session.info.pop(_FINALIZING_KEY, None)
+
+
+def _reset_after_outer_transaction(
+    session: Session, transaction: SessionTransaction
+) -> None:
+    """Clear retained state only when the outer transaction has ended."""
+    if transaction.parent is None:
+        _reset_transaction_state(session)
 
 
 def _append_child_records_to_buffer(
@@ -231,8 +232,8 @@ def _append_child_records_to_buffer(
 ) -> None:
     """Compute dataset + dashboard child-collection records + append to buffer.
 
-    Runs in ``after_flush`` so the shadow tables already have the
-    current-tx rows. Reads from Continuum shadow tables
+    Runs during commit finalization after the explicit final flush, so the
+    shadow tables already have the current-tx rows. Reads from Continuum shadow tables
     (``table_columns_version`` / ``sql_metrics_version`` /
     ``dashboard_slices_version`` / ``slices_version``).
     """
@@ -280,17 +281,10 @@ def _inject_action_meta_record(
     """Pop ``ACTION_META_KEY`` and prepend its synthetic headline record
     to the owning entity's buffer (the ``__meta__`` record convention).
 
-    No-op when no command set the key — and, critically, no-op WITHOUT
-    popping when the buffer is empty: the buffer-empty short-circuit in
-    ``flush_change_records`` exists so a multi-flush transaction can
-    deliver its records on a later firing, and a headline-only buffer
-    would defeat it (the first firing would persist just the headline,
-    mark the tx processed, and the later flush's real records would be
-    silently dropped). Leaving the key in place parks the headline until
-    the record-bearing firing. Prepended (not appended) so the headline
-    gets ``sequence`` 0 and renders first. Malformed payloads are logged
-    and dropped — a headline is descriptive enrichment, never worth
-    failing the user's save over.
+    No-op when no command set the key and, critically, does not pop when the
+    final buffer is empty. Prepended rather than appended so the headline gets
+    ``sequence`` 0 and renders first. Malformed payloads are logged and dropped:
+    a headline is descriptive enrichment, never worth failing the user's save.
     """
     if not buffer:
         return
@@ -379,7 +373,7 @@ def _persist_buffered_records(
 
 
 def register_change_record_listener() -> None:  # noqa: C901
-    """Attach the before_flush + after_flush listeners.
+    """Attach transaction-scoped version-change listeners.
 
     Registered from :class:`superset.initialization.SupersetAppInitializer`
     (``init_versioning``) alongside the baseline, dataset-snapshot,
@@ -398,110 +392,42 @@ def register_change_record_listener() -> None:  # noqa: C901
 
     versioned_classes: tuple[type, ...] = (Dashboard, Slice, SqlaTable)
 
-    def compute_change_records(
+    def capture_initial_states(
         session: Session, _flush_context: Any, _instances: Any
     ) -> None:
-        # session.info persists across before_flush/after_flush within
-        # a single transaction. The buffer is keyed on
-        # ``(entity_kind, entity_id)`` so scalar records captured here
-        # and child records captured in after_flush merge
-        # under the same entity without duplication.
-        buffer: dict[tuple[str, int], list[ChangeRecord]] = session.info.setdefault(
-            _BUFFER_KEY, {}
+        initial_states: dict[tuple[str, int], tuple[Any, dict[str, Any]]] = (
+            session.info.setdefault(_INITIAL_STATES_KEY, {})
         )
         for obj in list(session.dirty):
             if isinstance(obj, versioned_classes):
-                _process_dirty_entity_into_buffer(session, obj, buffer)
+                _capture_dirty_entity_initial_state(session, obj, initial_states)
 
-    def flush_change_records(session: Session, _flush_context: Any) -> None:
-        buffer: dict[tuple[str, int], list[ChangeRecord]] = session.info.setdefault(
-            _BUFFER_KEY, {}
-        )
-
-        tx_id = _current_transaction_id(session)
-        if tx_id is None:
-            session.info[_BUFFER_KEY] = {}
-            return
-
-        # Skip if we've already written records for this tx (after_flush
-        # can fire more than once per commit — e.g. autoflush from a
-        # mid-commit query). Without this guard the child-diff path would
-        # re-read the same shadow rows and re-emit the same records,
-        # tripping the UNIQUE(transaction_id, entity_kind, entity_id,
-        # sequence) constraint on insert.
-        processed: set[int] = session.info.setdefault(_PROCESSED_TXS_KEY, set())
-        if tx_id in processed:
-            # Drop anything buffered after the tx was persisted: records
-            # left here would otherwise survive on the long-lived scoped
-            # session and be inserted under the NEXT transaction's id.
-            session.info[_BUFFER_KEY] = {}
-            return
-
-        # Stamp action_kind eagerly, before the buffer-empty short-
-        # circuit. Restores / imports / clones may flush across multiple
-        # cycles; the FIRST firing for this tx is the one with the
-        # value still on ``session.info``. The helper pops on success
-        # so subsequent firings see ``None`` and short-circuit cleanly.
-        _stamp_action_kind_on_transaction(session, tx_id)
-
-        _append_child_records_to_buffer(session, tx_id, buffer)
-
-        # After the child append and before the emptiness check: the
-        # headline joins whichever firing carries the transaction's real
-        # records (scalar or child), and its peek-don't-pop guard parks
-        # it across record-less firings instead of defeating the
-        # multi-flush short-circuit below.
-        _inject_action_meta_record(session, buffer)
-
-        if not buffer:
-            # Don't mark tx as processed when nothing was inserted. A
-            # later after_flush firing for the same tx may carry the
-            # records — e.g. when an entity's edit lands across two
-            # flushes (a child-only flush followed by a parent-dirty
-            # flush): the parent shadow only lands in the parent-dirty
-            # flush, so the child-diff path can't find a prior tx to
-            # compare against until then.
-            session.info[_BUFFER_KEY] = {}
+    def finalize_change_records(session: Session) -> None:
+        if session.in_nested_transaction() or session.info.get(_FINALIZING_KEY):
             return
 
+        session.info[_FINALIZING_KEY] = True
         try:
-            _persist_buffered_records(session, tx_id, buffer)
+            session.flush()
+            initial_states: dict[tuple[str, int], tuple[Any, dict[str, Any]]] = (
+                session.info.get(_INITIAL_STATES_KEY, {})
+            )
+            buffer = _build_scalar_buffer(initial_states)
+
+            tx_id = _current_transaction_id(session)
+            if tx_id is None:
+                return
+
+            _stamp_action_kind_on_transaction(session, tx_id)
+            _append_child_records_to_buffer(session, tx_id, buffer)
+            _inject_action_meta_record(session, buffer)
+
+            if buffer:
+                _persist_buffered_records(session, tx_id, buffer)
         finally:
-            session.info[_BUFFER_KEY] = {}
-            processed.add(tx_id)
+            session.info.pop(_FINALIZING_KEY, None)
 
-    def reset_processed_after_commit(session: Session) -> None:
-        # ``_PROCESSED_TXS_KEY`` accumulates Continuum tx ids whose change
-        # records have already been written, to dedup against multiple
-        # ``after_flush`` firings within one transaction. After commit
-        # the tx is closed and its id will never recur on this session
-        # — drop the set so a long-lived session (Celery worker, CLI)
-        # doesn't grow it without bound.
-        session.info.pop(_PROCESSED_TXS_KEY, None)
-        # If a command set the action_kind but no flush fired (e.g. a
-        # save that touched nothing versioned), the value would
-        # otherwise leak into the next transaction. Drop it here as a
-        # belt-and-suspenders cleanup; the
-        # ``_stamp_action_kind_on_transaction`` helper already pops on
-        # the normal path.
-        session.info.pop(ACTION_KIND_KEY, None)
-        session.info.pop(ACTION_META_KEY, None)
-        session.info.pop(_BUFFER_KEY, None)
-
-    def reset_action_kind_after_rollback(session: Session) -> None:
-        # When a command sets ``ACTION_KIND_KEY`` and then an exception
-        # fires before flush (e.g. validation error after the key is
-        # set), the transaction rolls back without the listener ever
-        # popping the key. The next save on the same session would
-        # then inherit the stale value and label an unrelated commit
-        # as "restore" / "import" / "clone". Pop here so a rolled-back
-        # action's intent doesn't leak forward.
-        session.info.pop(ACTION_KIND_KEY, None)
-        session.info.pop(ACTION_META_KEY, None)
-        session.info.pop(_BUFFER_KEY, None)
-
-    event.listen(db.session, "before_flush", compute_change_records)
-    event.listen(db.session, "after_flush", flush_change_records)
-    event.listen(db.session, "after_commit", reset_processed_after_commit)
-    event.listen(db.session, "after_rollback", reset_action_kind_after_rollback)
+    event.listen(db.session, "before_flush", capture_initial_states)
+    event.listen(db.session, "before_commit", finalize_change_records)
+    event.listen(db.session, "after_transaction_end", _reset_after_outer_transaction)
     setattr(db.session, _REGISTERED_SENTINEL, True)

superset/versioning/changes/state.py

@@ -24,7 +24,7 @@ Three concerns live here:
 2. **State capture** — :func:`_orm_to_post_state` serialises the
    in-memory ORM object; :func:`_read_pre_state` reads the corresponding
    pre-flush row directly from the DB inside ``session.no_autoflush``.
-3. **Diff dispatch** — :func:`compute_records_for_entity` routes to the
+3. **Diff dispatch** — :func:`compute_records_from_state` routes to the
    right :mod:`superset.versioning.diff` helper based on the model
    class name (string dispatch keeps this module free of hard imports
    on the three entity classes, which avoids import-order coupling at
@@ -156,36 +156,29 @@ def _read_pre_state(
     return {key: jsonable(value) for key, value in result.items()}
 
 
-def compute_records_for_entity(session: Session, obj: Any) -> list[ChangeRecord]:
-    """Diff the pre-state (from DB) against the post-state (in memory).
-
-    Dispatches to :func:`diff_slice` / :func:`diff_dashboard` /
-    :func:`diff_dataset` based on the model class name — string-based
-    dispatch is used to keep this module free of hard imports on the
-    three entity classes, which in turn avoids import-order coupling
-    at app-init time.
-    """
-    model_cls = type(obj)
+def capture_initial_state(session: Session, obj: Any) -> dict[str, Any] | None:
+    """Capture an entity's database state before its first transaction flush."""
     entity_id = getattr(obj, "id", None)
     if entity_id is None:
-        return []
-
+        return None
     try:
-        pre_state = _read_pre_state(session, model_cls, entity_id)
+        return _read_pre_state(session, type(obj), entity_id)
     except Exception:  # pylint: disable=broad-except
         logger.exception(
-            "version_changes: pre-state read failed for %s id=%s",
-            model_cls.__name__,
+            "version_changes: initial-state read failed for %s id=%s",
+            type(obj).__name__,
             entity_id,
         )
-        return []
+        return None
 
-    if pre_state is None:
-        return []
 
+def compute_records_from_state(
+    obj: Any, pre_state: dict[str, Any]
+) -> list[ChangeRecord]:
+    """Diff a retained transaction pre-state against an entity's final state."""
     post_state = _orm_to_post_state(obj)
+    model_cls = type(obj)
     fields = _cached_scalar_fields(model_cls)
-
     name = model_cls.__name__
     if name == "Slice":
         return diff_slice(pre_state, post_state, fields=fields)

tests/integration_tests/versioning/transaction_correctness_tests.py

@@ -0,0 +1,224 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Transaction-level correctness tests for entity version history."""
+
+import pytest
+import sqlalchemy as sa
+
+from superset import db
+from superset.connectors.sqla.models import SqlaTable
+from superset.models.slice import Slice
+from superset.utils import json
+from superset.versioning.changes.table import version_changes_table
+from tests.integration_tests.base_tests import SupersetTestCase
+from tests.integration_tests.fixtures.birth_names_dashboard import (  # noqa: F401
+    load_birth_names_dashboard_with_slices,
+    load_birth_names_data,
+)
+
+
+def _latest_transaction_id() -> int:
+    """Return the latest semantic-history transaction boundary."""
+    return db.session.scalar(
+        sa.select(
+            sa.func.coalesce(sa.func.max(version_changes_table.c.transaction_id), 0)
+        )
+    )
+
+
+def _changes_for_chart(
+    chart_id: int, *, after_transaction_id: int
+) -> list[dict[str, object]]:
+    """Return ordered semantic changes captured for one chart."""
+    rows = db.session.execute(
+        sa.select(version_changes_table)
+        .where(version_changes_table.c.entity_kind == "chart")
+        .where(version_changes_table.c.entity_id == chart_id)
+        .where(version_changes_table.c.transaction_id > after_transaction_id)
+        .order_by(
+            version_changes_table.c.transaction_id,
+            version_changes_table.c.sequence,
+        )
+    ).mappings()
+    return [dict(row) for row in rows]
+
+
+def _changes_for_dataset(
+    dataset_id: int, *, after_transaction_id: int
+) -> list[dict[str, object]]:
+    """Return ordered semantic changes captured for one dataset."""
+    rows = db.session.execute(
+        sa.select(version_changes_table)
+        .where(version_changes_table.c.entity_kind == "dataset")
+        .where(version_changes_table.c.entity_id == dataset_id)
+        .where(version_changes_table.c.transaction_id > after_transaction_id)
+        .order_by(
+            version_changes_table.c.transaction_id,
+            version_changes_table.c.sequence,
+        )
+    ).mappings()
+    return [dict(row) for row in rows]
+
+
+@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
+class TestVersionHistoryTransactionCorrectness(SupersetTestCase):
+    """Verify semantic projections span every flush in one transaction."""
+
+    @staticmethod
+    def _chart(name: str) -> Slice:
+        chart = db.session.query(Slice).filter(Slice.slice_name == name).one()
+        db.session.commit()
+        return chart
+
+    def test_same_entity_multiple_flushes_produce_one_net_change(self) -> None:
+        """Only the initial-to-final scalar transition is materialized."""
+        chart = self._chart("Girls")
+        chart_id = chart.id
+        boundary = _latest_transaction_id()
+        try:
+            chart.slice_name = "Girls intermediate"
+            db.session.flush()
+            chart.slice_name = "Girls final"
+            db.session.commit()
+
+            changes = _changes_for_chart(chart_id, after_transaction_id=boundary)
+            name_changes = [row for row in changes if row["path"] == ["slice_name"]]
+            assert len(name_changes) == 1
+            assert name_changes[0]["from_value"] == "Girls"
+            assert name_changes[0]["to_value"] == "Girls final"
+        finally:
+            chart = db.session.get(Slice, chart_id)
+            assert chart is not None
+            chart.slice_name = "Girls"
+            db.session.commit()
+
+    def test_different_entities_across_flushes_share_complete_history(self) -> None:
+        """A later entity flush is not discarded from the transaction."""
+        girls = self._chart("Girls")
+        boys = self._chart("Boys")
+        boundary = _latest_transaction_id()
+        try:
+            girls.slice_name = "Girls transaction edit"
+            db.session.flush()
+            boys.slice_name = "Boys transaction edit"
+            db.session.commit()
+
+            girls_changes = [
+                row
+                for row in _changes_for_chart(girls.id, after_transaction_id=boundary)
+                if row["to_value"] == "Girls transaction edit"
+            ]
+            boys_changes = [
+                row
+                for row in _changes_for_chart(boys.id, after_transaction_id=boundary)
+                if row["to_value"] == "Boys transaction edit"
+            ]
+            assert len(girls_changes) == 1
+            assert len(boys_changes) == 1
+            assert (
+                girls_changes[0]["transaction_id"] == boys_changes[0]["transaction_id"]
+            )
+        finally:
+            girls.slice_name = "Girls"
+            boys.slice_name = "Boys"
+            db.session.commit()
+
+    def test_return_to_initial_state_produces_no_duplicate_change(self) -> None:
+        """Intermediate-only edits disappear from the net projection."""
+        chart = self._chart("Girls")
+        boundary = _latest_transaction_id()
+        chart.slice_name = "Girls temporary"
+        db.session.flush()
+        chart.slice_name = "Girls"
+        db.session.commit()
+
+        assert not [
+            row
+            for row in _changes_for_chart(chart.id, after_transaction_id=boundary)
+            if row["path"] == ["slice_name"]
+        ]
+
+    def test_child_changes_across_flushes_use_final_shadow_state(self) -> None:
+        """A child-only edit projects the final child state on its parent."""
+        dataset = (
+            db.session.query(SqlaTable)
+            .filter(SqlaTable.table_name == "birth_names")
+            .first()
+        )
+        assert dataset is not None
+        assert dataset.columns
+        column = dataset.columns[0]
+        original = column.description
+        db.session.commit()
+        boundary = _latest_transaction_id()
+        try:
+            column.description = "intermediate child description"
+            db.session.flush()
+            column.description = "final child description"
+            db.session.commit()
+
+            serialized = json.dumps(
+                _changes_for_dataset(dataset.id, after_transaction_id=boundary)
+            )
+            assert "final child description" in serialized
+            assert "intermediate child description" not in serialized
+        finally:
+            column.description = original
+            db.session.commit()
+
+    def _assert_nested_transaction_preserves_outer_initial_state(
+        self, nested_outcome: str
+    ) -> None:
+        """SAVEPOINT completion does not clear the outer transaction registry."""
+        chart = self._chart("Girls")
+        chart_id = chart.id
+        boundary = _latest_transaction_id()
+        try:
+            chart.slice_name = "Girls outer edit"
+            db.session.flush()
+
+            nested = db.session.begin_nested()
+            chart.slice_name = "Girls nested edit"
+            db.session.flush()
+            getattr(nested, nested_outcome)()
+            db.session.commit()
+
+            changes = [
+                row
+                for row in _changes_for_chart(chart_id, after_transaction_id=boundary)
+                if row["path"] == ["slice_name"]
+            ]
+            assert len(changes) == 1
+            assert changes[0]["from_value"] == "Girls"
+            assert changes[0]["to_value"] == (
+                "Girls nested edit"
+                if nested_outcome == "commit"
+                else "Girls outer edit"
+            )
+        finally:
+            chart = db.session.get(Slice, chart_id)
+            assert chart is not None
+            chart.slice_name = "Girls"
+            db.session.commit()
+
+    def test_nested_commit_preserves_outer_initial_state(self) -> None:
+        """Committing a SAVEPOINT keeps the outer transaction registry."""
+        self._assert_nested_transaction_preserves_outer_initial_state("commit")
+
+    def test_nested_rollback_preserves_outer_initial_state(self) -> None:
+        """Rolling back a SAVEPOINT keeps the outer transaction registry."""
+        self._assert_nested_transaction_preserves_outer_initial_state("rollback")

tests/unit_tests/commands/dataset/test_duplicate.py

@@ -456,6 +456,8 @@ def test_duplicate_dataset_with_columns_and_metrics() -> None:
     mock_column.is_dttm = False
     mock_column.type = "VARCHAR"
     mock_column.description = "Test column"
+    mock_column.filterable = False
+    mock_column.groupby = False
 
     mock_metric = Mock(spec=SqlMetric)
     mock_metric.metric_name = "count"
@@ -463,6 +465,10 @@ def test_duplicate_dataset_with_columns_and_metrics() -> None:
     mock_metric.expression = "COUNT(*)"
     mock_metric.metric_type = "count"
     mock_metric.description = "Row count"
+    mock_metric.d3format = ",.2f"
+    mock_metric.currency = None
+    mock_metric.warning_text = "approximate"
+    mock_metric.extra = None
 
     mock_base_model = Mock(spec=SqlaTable)
     mock_base_model.id = 1
@@ -518,7 +524,14 @@ def test_duplicate_dataset_with_columns_and_metrics() -> None:
                         # Verify columns were duplicated
                         assert len(result.columns) == 1
                         assert result.columns[0].column_name == "col1"
+                        # Non-default groupby/filterable flags must survive the
+                        # clone rather than being reset to the column defaults.
+                        assert result.columns[0].filterable is False
+                        assert result.columns[0].groupby is False
 
                         # Verify metrics were duplicated
                         assert len(result.metrics) == 1
                         assert result.metrics[0].metric_name == "count"
+                        # Formatting/metadata fields must survive the clone too.
+                        assert result.metrics[0].d3format == ",.2f"
+                        assert result.metrics[0].warning_text == "approximate"

tests/unit_tests/commands/importers/v1/utils_test.py

@@ -16,10 +16,17 @@
 # under the License.
 """Tests for superset/commands/dataset/importers/v1/utils.py temporal helpers."""
 
+import logging
 from unittest.mock import patch
 
 import pandas as pd
 import pytest
+from jsonpath_ng import parse
+from marshmallow import Schema
+from marshmallow.exceptions import ValidationError
+
+from superset.constants import PASSWORD_MASK
+from superset.utils import json
 
 
 class TestConvertTemporalColumns:
@@ -117,3 +124,62 @@ class TestConvertTemporalColumns:
 
         call_args = mock_logger.warning.call_args[0]
         assert call_args[1] == 2  # 2 out-of-bounds, 1 pre-existing null
+
+
+class TestLoadConfigs:
+    def test_load_configs_caught_exception_and_log_redacted_configs(self) -> None:
+        logging.basicConfig(level=logging.DEBUG)
+        from superset.commands.importers.v1.utils import (
+            load_configs,
+        )
+
+        with (
+            patch("superset.db") as mock_db,
+            patch("yaml.safe_load") as mock_yaml_safe_load,
+            patch("superset.commands.importers.v1.utils.logger") as mock_logger,
+        ):
+            mock_yaml_safe_load.return_value = {
+                "masked_encrypted_extra": json.dumps(
+                    {"oauth2_client_info": {"secret": "MASKED"}}
+                ),
+                "ssh_tunnel": {},
+            }
+            mock_db.session.query.return_value = {"uuid-1": "SECRET_TO_BE_SEEN"}
+
+            class RaiseValidationSchema(Schema):
+                def load(self, value):
+                    raise ValidationError("Exception when validating config")
+
+            failed_schema = RaiseValidationSchema()
+
+            load_configs(
+                contents={"file1/": "here is some content"},
+                schemas={
+                    "file1/": failed_schema,
+                },
+                passwords={"file1/": "PASSWORD"},
+                exceptions=[],
+                ssh_tunnel_passwords={"file1/": "SECRET_SSH_TUNNEL_PASSWORD"},
+                ssh_tunnel_private_keys={"file1/": "SECRET_SSH_TUNNEL_PRIVATE_KEY"},
+                ssh_tunnel_priv_key_passwords={
+                    "file1/": "SECRET_SSH_TUNNEL_PRIV_KEY_PASSWORD"
+                },
+                encrypted_extra_secrets={
+                    "file1/": {
+                        "$.oauth2_client_info.secret": "SECRET_OAUTH2_CLIENT_INFO"
+                    }
+                },
+            )
+
+            logged_config = mock_logger.debug.call_args[0][1]
+
+            for redacted_field in [
+                "$.password",
+                "$.ssh_tunnel.password",
+                "$.ssh_tunnel.private_key",
+                "$.ssh_tunnel.private_key_password",
+                "$.masked_encrypted_extra",
+            ]:
+                jsonpath_expr = parse(redacted_field)
+                for match in jsonpath_expr.find(logged_config):
+                    assert match.value == PASSWORD_MASK

tests/unit_tests/models/helpers_test.py

@@ -32,7 +32,7 @@ from sqlalchemy.pool import StaticPool
 from sqlalchemy.sql.elements import ColumnElement
 
 from superset.superset_typing import AdhocColumn, AdhocMetric, OrderBy
-from superset.utils.core import GenericDataType
+from superset.utils.core import FilterOperator, GenericDataType
 from tests.unit_tests.conftest import with_feature_flags
 
 if TYPE_CHECKING:
@@ -3410,3 +3410,265 @@ def test_get_sqla_query_dotted_struct_column_bigquery(
     # ```forecasts.original`.`total_cost``` (the regression), so this negative
     # assertion catches the actual failure mode, not just an exact-string match.
     assert "`forecasts.original`" not in sql
+
+
+def test_temporal_epoch_string_filter_is_coerced_for_bigquery() -> None:
+    """
+    Drill-to-detail can send JavaScript timestamp strings for temporal values.
+    BigQuery DATE filters must receive a DATE literal instead of the raw string.
+    """
+    from superset.db_engine_specs.bigquery import BigQueryEngineSpec
+    from superset.models.helpers import ExploreMixin
+
+    value = ExploreMixin.filter_values_handler(
+        values="1778630400000",
+        operator=FilterOperator.EQUALS,
+        target_generic_type=GenericDataType.TEMPORAL,
+        target_native_type="DATE",
+        db_engine_spec=BigQueryEngineSpec,
+    )
+
+    assert isinstance(value, ColumnElement)
+    assert str(value) == "CAST('2026-05-13' AS DATE)"
+
+
+def test_temporal_negative_epoch_string_filter_is_coerced_for_bigquery() -> None:
+    from superset.db_engine_specs.bigquery import BigQueryEngineSpec
+    from superset.models.helpers import ExploreMixin
+
+    value = ExploreMixin.filter_values_handler(
+        values="-1778630400000",
+        operator=FilterOperator.EQUALS,
+        target_generic_type=GenericDataType.TEMPORAL,
+        target_native_type="DATE",
+        db_engine_spec=BigQueryEngineSpec,
+    )
+
+    assert isinstance(value, ColumnElement)
+    assert str(value) == "CAST('1913-08-22' AS DATE)"
+
+
+@pytest.mark.parametrize("value", [1778630400000, 1778630400000.0])
+def test_temporal_numeric_filter_is_coerced_for_bigquery(
+    value: int | float,
+) -> None:
+    from superset.db_engine_specs.bigquery import BigQueryEngineSpec
+    from superset.models.helpers import ExploreMixin
+
+    result = ExploreMixin.filter_values_handler(
+        values=value,
+        operator=FilterOperator.EQUALS,
+        target_generic_type=GenericDataType.TEMPORAL,
+        target_native_type="DATE",
+        db_engine_spec=BigQueryEngineSpec,
+    )
+
+    assert isinstance(result, ColumnElement)
+    assert str(result) == "CAST('2026-05-13' AS DATE)"
+
+
+def test_temporal_overflow_epoch_filter_is_not_coerced() -> None:
+    from superset.db_engine_specs.bigquery import BigQueryEngineSpec
+    from superset.models.helpers import ExploreMixin
+
+    value = "999999999999999999999999999999999999999"
+
+    result = ExploreMixin.filter_values_handler(
+        values=value,
+        operator=FilterOperator.EQUALS,
+        target_generic_type=GenericDataType.TEMPORAL,
+        target_native_type="DATE",
+        db_engine_spec=BigQueryEngineSpec,
+    )
+
+    assert result == value
+
+
+def test_temporal_epoch_filter_is_not_coerced_without_engine_literal() -> None:
+    from superset.db_engine_specs.base import BaseEngineSpec
+    from superset.models.helpers import ExploreMixin
+
+    result = ExploreMixin.filter_values_handler(
+        values="1778630400000",
+        operator=FilterOperator.EQUALS,
+        target_generic_type=GenericDataType.TEMPORAL,
+        target_native_type="UNKNOWN_TYPE",
+        db_engine_spec=BaseEngineSpec,
+    )
+
+    assert result == "1778630400000"
+
+
+@pytest.mark.parametrize(
+    "operator",
+    [FilterOperator.IN, FilterOperator.NOT_IN],
+)
+def test_temporal_epoch_string_filter_list_is_coerced_for_bigquery(
+    operator: FilterOperator,
+) -> None:
+    """
+    Cross-filtering can send JavaScript timestamp strings inside IN lists.
+    BigQuery DATE filters must receive DATE literals instead of raw strings.
+    """
+    from superset.db_engine_specs.bigquery import BigQueryEngineSpec
+    from superset.models.helpers import ExploreMixin
+
+    values = ExploreMixin.filter_values_handler(
+        values=["1777248000000"],
+        operator=operator,
+        target_generic_type=GenericDataType.TEMPORAL,
+        target_native_type="DATE",
+        is_list_target=True,
+        db_engine_spec=BigQueryEngineSpec,
+    )
+
+    assert isinstance(values, list)
+    assert len(values) == 1
+    assert isinstance(values[0], ColumnElement)
+    assert str(values[0]) == "CAST('2026-04-27' AS DATE)"
+
+
+def test_temporal_epoch_string_filter_list_is_coerced_for_clickhouse() -> None:
+    from superset.db_engine_specs.clickhouse import ClickHouseEngineSpec
+    from superset.models.helpers import ExploreMixin
+
+    values = ExploreMixin.filter_values_handler(
+        values=["1777248000000"],
+        operator=FilterOperator.IN,
+        target_generic_type=GenericDataType.TEMPORAL,
+        target_native_type="Date",
+        is_list_target=True,
+        db_engine_spec=ClickHouseEngineSpec,
+    )
+
+    assert isinstance(values, list)
+    assert len(values) == 1
+    assert isinstance(values[0], ColumnElement)
+    assert str(values[0]) == "toDate('2026-04-27')"
+
+
+@pytest.mark.parametrize(
+    "target_type,expected",
+    [
+        ("DATE", "TO_DATE('2026-05-13', 'YYYY-MM-DD')"),
+        (
+            "TIMESTAMP",
+            "TO_TIMESTAMP('2026-05-13 00:00:00.000000', 'YYYY-MM-DD HH24:MI:SS.US')",
+        ),
+    ],
+)
+def test_temporal_epoch_string_filter_is_coerced_for_postgres(
+    target_type: str,
+    expected: str,
+) -> None:
+    from superset.db_engine_specs.postgres import PostgresEngineSpec
+    from superset.models.helpers import ExploreMixin
+
+    value = ExploreMixin.filter_values_handler(
+        values="1778630400000",
+        operator=FilterOperator.EQUALS,
+        target_generic_type=GenericDataType.TEMPORAL,
+        target_native_type=target_type,
+        db_engine_spec=PostgresEngineSpec,
+    )
+
+    assert isinstance(value, ColumnElement)
+    assert str(value) == expected
+
+
+def test_temporal_epoch_string_filter_list_is_coerced_for_postgres() -> None:
+    from superset.db_engine_specs.postgres import PostgresEngineSpec
+    from superset.models.helpers import ExploreMixin
+
+    values = ExploreMixin.filter_values_handler(
+        values=["1777248000000"],
+        operator=FilterOperator.IN,
+        target_generic_type=GenericDataType.TEMPORAL,
+        target_native_type="DATE",
+        is_list_target=True,
+        db_engine_spec=PostgresEngineSpec,
+    )
+
+    assert isinstance(values, list)
+    assert len(values) == 1
+    assert isinstance(values[0], ColumnElement)
+    assert str(values[0]) == "TO_DATE('2026-04-27', 'YYYY-MM-DD')"
+
+
+def test_non_temporal_numeric_string_filter_is_not_coerced() -> None:
+    from superset.db_engine_specs.bigquery import BigQueryEngineSpec
+    from superset.models.helpers import ExploreMixin
+
+    value = ExploreMixin.filter_values_handler(
+        values="1778630400000",
+        operator=FilterOperator.EQUALS,
+        target_generic_type=GenericDataType.STRING,
+        target_native_type="DATE",
+        db_engine_spec=BigQueryEngineSpec,
+    )
+
+    assert value == "1778630400000"
+
+
+def test_temporal_range_filter_value_is_not_coerced() -> None:
+    from superset.db_engine_specs.bigquery import BigQueryEngineSpec
+    from superset.models.helpers import ExploreMixin
+
+    value = ExploreMixin.filter_values_handler(
+        values="No filter",
+        operator=FilterOperator.TEMPORAL_RANGE,
+        target_generic_type=GenericDataType.TEMPORAL,
+        target_native_type="DATE",
+        db_engine_spec=BigQueryEngineSpec,
+    )
+
+    assert value == "No filter"
+
+
+@pytest.mark.parametrize(
+    "target_type,expected",
+    [
+        (
+            "DATETIME",
+            "CAST('2026-05-13T14:30:00.123000' AS DATETIME)",
+        ),
+        (
+            "TIMESTAMP",
+            "CAST('2026-05-13T14:30:00.123000' AS TIMESTAMP)",
+        ),
+    ],
+)
+def test_temporal_epoch_string_filter_is_coerced_for_datetime_bigquery(
+    target_type: str,
+    expected: str,
+) -> None:
+    """Sub-second epoch precision survives DATETIME / TIMESTAMP coercion."""
+    from superset.db_engine_specs.bigquery import BigQueryEngineSpec
+    from superset.models.helpers import ExploreMixin
+
+    value = ExploreMixin.filter_values_handler(
+        values="1778682600123",
+        operator=FilterOperator.EQUALS,
+        target_generic_type=GenericDataType.TEMPORAL,
+        target_native_type=target_type,
+        db_engine_spec=BigQueryEngineSpec,
+    )
+
+    assert isinstance(value, ColumnElement)
+    assert str(value) == expected
+
+
+def test_temporal_non_numeric_string_filter_is_not_coerced() -> None:
+    """Non-integer temporal strings pass through without epoch coercion."""
+    from superset.db_engine_specs.bigquery import BigQueryEngineSpec
+    from superset.models.helpers import ExploreMixin
+
+    value = ExploreMixin.filter_values_handler(
+        values="2025-12-20",
+        operator=FilterOperator.EQUALS,
+        target_generic_type=GenericDataType.TEMPORAL,
+        target_native_type="DATE",
+        db_engine_spec=BigQueryEngineSpec,
+    )
+
+    assert value == "2025-12-20"

tests/unit_tests/utils/csv_tests.py

@@ -67,6 +67,9 @@ def test_escape_value():
     result = csv.escape_value(" =10+2")
     assert result == "' =10+2"
 
+    result = csv.escape_value('  ""=10+2')
+    assert result == '\'  ""=10+2'
+
     # A leading tab or carriage return followed by a dangerous char was already
     # handled by \s{1,} in the pre-existing regex. The cases below test the
     # new behavior: tab/CR alone (not followed by a dangerous char) are now

tests/unit_tests/utils/test_date_parsing.py

@@ -17,11 +17,14 @@
 """Tests for datetime format detection and warning suppression."""
 
 import warnings
+from datetime import datetime
 
 import pandas as pd
 import pytest
+import pytz
 
 from superset.utils.core import DateColumn, normalize_dttm_col
+from superset.utils.dates import datetime_to_epoch
 from superset.utils.pandas import detect_datetime_format
 
 
@@ -254,3 +257,91 @@ def test_warning_suppression():
 
     assert len(warnings_list) == 0  # Should suppress all format inference warnings
     assert pd.api.types.is_datetime64_any_dtype(df["date"])  # Should still parse dates
+
+
+# ============================================================================
+# NEW TESTS FOR datetime_to_epoch() - Edge case coverage
+# ============================================================================
+
+
+def test_datetime_to_epoch_naive_at_epoch():
+    """Test naive datetime exactly at epoch returns 0.0"""
+    # Edge case: Datetime at epoch boundary
+    epoch_dt = datetime(1970, 1, 1, 0, 0, 0)
+    result = datetime_to_epoch(epoch_dt)
+    assert result == 0.0, f"Epoch datetime should be 0.0, got {result}"
+
+
+def test_datetime_to_epoch_naive_one_second_after():
+    """Test naive datetime 1 second after epoch"""
+    dt = datetime(1970, 1, 1, 0, 0, 1)
+    result = datetime_to_epoch(dt)
+    expected = 1000.0  # 1 second * 1000 ms
+    assert result == expected, f"Expected {expected}ms, got {result}ms"
+
+
+def test_datetime_to_epoch_timezone_aware_utc():
+    """Test timezone-aware datetime in UTC"""
+    # Create UTC datetime
+    utc_tz = pytz.UTC
+    dt_utc = utc_tz.localize(datetime(1970, 1, 1, 0, 0, 1))
+    result = datetime_to_epoch(dt_utc)
+    expected = 1000.0  # 1 second * 1000 ms
+    assert result == expected, f"UTC datetime should convert correctly, got {result}ms"
+
+
+def test_datetime_to_epoch_timezone_aware_different_tz():
+    """Test timezone-aware datetime in different timezone converts to UTC correctly"""
+    # Create datetime in EST (UTC-5 in January)
+    est = pytz.timezone("US/Eastern")
+    # 1970-01-01 05:00:00 EST = 1970-01-01 10:00:00 UTC (5 hours offset)
+    dt_est = est.localize(datetime(1970, 1, 1, 5, 0, 0))
+    result = datetime_to_epoch(dt_est)
+    expected = 10 * 60 * 60 * 1000  # 10 hours in milliseconds
+    assert result == expected, (
+        f"EST datetime should convert to UTC correctly, got {result}ms"
+    )
+
+
+def test_datetime_to_epoch_dst_transition():
+    """Test datetime during DST transition is handled correctly"""
+    # Use a known DST transition date in US/Eastern
+    # 2023-03-12: Spring forward (2 AM becomes 3 AM, gap of 1 hour)
+    eastern = pytz.timezone("US/Eastern")
+
+    # Create datetime before DST transition (still EST, standard time)
+    dt_before_dst = eastern.localize(datetime(2023, 3, 12, 1, 59, 59), is_dst=False)
+    result_before = datetime_to_epoch(dt_before_dst)
+
+    # Create datetime after DST transition (now EDT, daylight time)
+    dt_after_dst = eastern.localize(datetime(2023, 3, 12, 3, 0, 1), is_dst=True)
+    result_after = datetime_to_epoch(dt_after_dst)
+
+    # The difference should be exactly 2 seconds, not 1 hour + 2 seconds
+    # (because of the DST jump, 1:59:59 EST -> 3:00:01 EDT)
+    diff_ms = result_after - result_before
+    expected_diff = 2000  # 2 seconds
+    assert diff_ms == expected_diff, (
+        f"DST transition handled incorrectly. Diff: {diff_ms}ms"
+    )
+
+
+def test_datetime_to_epoch_microsecond_precision():
+    """Test that microseconds are handled correctly"""
+    dt = datetime(1970, 1, 1, 0, 0, 1, 500000)  # 1.5 seconds
+    result = datetime_to_epoch(dt)
+    expected = 1500.0  # 1.5 seconds * 1000 ms
+    assert result == expected, (
+        f"Microseconds should contribute to result, got {result}ms"
+    )
+
+
+def test_datetime_to_epoch_far_future():
+    """Test datetime far in the future"""
+    # 2050-01-01 should work without errors
+    dt = datetime(2050, 1, 1, 0, 0, 0)
+    result = datetime_to_epoch(dt)
+    # Just verify it's a reasonable large number (no crashes, reasonable value)
+    assert isinstance(result, float), "Should return float"
+    assert result > 0, "Far future date should have positive epoch"
+    assert result == 2524608000000.0, "2050-01-01 should be specific epoch value"

tests/unit_tests/versioning/test_listener.py

@@ -0,0 +1,204 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Tests for version-change listener transaction lifecycle behavior."""
+
+from collections.abc import Iterator
+from typing import Any
+
+import pytest
+import sqlalchemy as sa
+from sqlalchemy.orm import Session, sessionmaker
+
+from superset.versioning.changes import listener
+from superset.versioning.diff import ChangeRecord
+
+Base: Any = sa.orm.declarative_base()
+
+
+class LifecycleRow(Base):
+    """Minimal row used to characterize SQLAlchemy session events."""
+
+    __tablename__ = "versioning_listener_lifecycle"
+
+    id = sa.Column(sa.Integer, primary_key=True)
+    value = sa.Column(sa.String, nullable=False)
+
+
+@pytest.fixture
+def lifecycle_session() -> Iterator[Session]:
+    """Yield an isolated SQLAlchemy session backed by in-memory SQLite."""
+    engine = sa.create_engine("sqlite://")
+    Base.metadata.create_all(engine)
+    session = sessionmaker(bind=engine)()
+    try:
+        yield session
+    finally:
+        session.close()
+        engine.dispose()
+
+
+def test_before_commit_can_force_final_flush_without_reentry(
+    lifecycle_session: Session,
+) -> None:
+    """A before-commit finalizer can flush once before commit preparation."""
+    events: list[str] = []
+
+    @sa.event.listens_for(lifecycle_session, "before_flush")
+    def before_flush(
+        _session: Session, _flush_context: object, _instances: object
+    ) -> None:
+        events.append("before_flush")
+
+    @sa.event.listens_for(lifecycle_session, "after_flush")
+    def after_flush(_session: Session, _flush_context: object) -> None:
+        events.append("after_flush")
+
+    @sa.event.listens_for(lifecycle_session, "before_commit")
+    def before_commit(session: Session) -> None:
+        events.append("before_commit:start")
+        session.flush()
+        events.append("before_commit:end")
+
+    lifecycle_session.add(LifecycleRow(value="saved"))
+    lifecycle_session.commit()
+
+    assert events == [
+        "before_commit:start",
+        "before_flush",
+        "after_flush",
+        "before_commit:end",
+    ]
+
+
+def test_before_commit_with_no_work_does_not_flush(
+    lifecycle_session: Session,
+) -> None:
+    """A no-work commit invokes the finalizer without a flush cycle."""
+    events: list[str] = []
+
+    @sa.event.listens_for(lifecycle_session, "before_flush")
+    def before_flush(
+        _session: Session, _flush_context: object, _instances: object
+    ) -> None:
+        events.append("before_flush")
+
+    @sa.event.listens_for(lifecycle_session, "before_commit")
+    def before_commit(session: Session) -> None:
+        events.append("before_commit")
+        session.flush()
+
+    lifecycle_session.commit()
+
+    assert events == ["before_commit"]
+
+
+def test_rollback_event_can_clear_transaction_state(
+    lifecycle_session: Session,
+) -> None:
+    """Rollback cleanup runs while session-scoped state is still accessible."""
+    state_key = "_versioning_test_state"
+    lifecycle_session.info[state_key] = {"pending": True}
+
+    @sa.event.listens_for(lifecycle_session, "after_rollback")
+    def after_rollback(session: Session) -> None:
+        session.info.pop(state_key, None)
+
+    lifecycle_session.add(LifecycleRow(value="rolled back"))
+    lifecycle_session.flush()
+    lifecycle_session.rollback()
+
+    assert state_key not in lifecycle_session.info
+
+
+def test_capture_retains_the_first_pre_flush_state(
+    lifecycle_session: Session, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Repeated flushes retain the entity's initial database state once."""
+
+    class Slice:
+        id = 7
+
+    entity = Slice()
+    initial = {"slice_name": "initial"}
+    captures: list[object] = []
+
+    def capture(_session: Session, obj: object) -> dict[str, str]:
+        captures.append(obj)
+        return initial
+
+    monkeypatch.setattr(listener, "capture_initial_state", capture)
+    states: dict[tuple[str, int], tuple[object, dict[str, object]]] = {}
+
+    listener._capture_dirty_entity_initial_state(lifecycle_session, entity, states)
+    listener._capture_dirty_entity_initial_state(lifecycle_session, entity, states)
+
+    assert states == {("chart", 7): (entity, initial)}
+    assert captures == [entity]
+
+
+def test_scalar_buffer_materializes_one_final_net_diff(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Final materialization diffs each retained entity exactly once."""
+    entity = object()
+    initial = {"slice_name": "initial"}
+    record = ChangeRecord(
+        kind="property",
+        operation="edit",
+        path=["slice_name"],
+        from_value="initial",
+        to_value="final",
+    )
+    calls: list[tuple[object, dict[str, object]]] = []
+
+    def compute(obj: object, pre_state: dict[str, object]) -> list[ChangeRecord]:
+        calls.append((obj, pre_state))
+        return [record]
+
+    monkeypatch.setattr(listener, "compute_records_from_state", compute)
+
+    buffer = listener._build_scalar_buffer({("chart", 7): (entity, initial)})
+
+    assert buffer == {("chart", 7): [record]}
+    assert calls == [(entity, initial)]
+
+
+@pytest.mark.parametrize("terminal_event", ["commit", "rollback"])
+def test_terminal_event_clears_transaction_state(
+    lifecycle_session: Session, terminal_event: str
+) -> None:
+    """Commit and rollback cleanup discard all transaction-scoped state."""
+    lifecycle_session.info.update(
+        {
+            listener.ACTION_KIND_KEY: "restore",
+            listener.ACTION_META_KEY: {"headline": "restored"},
+            listener._INITIAL_STATES_KEY: {("chart", 7): object()},
+            listener._FINALIZING_KEY: True,
+            "unrelated": "preserved",
+        }
+    )
+    sa.event.listen(
+        lifecycle_session,
+        "after_transaction_end",
+        listener._reset_after_outer_transaction,
+    )
+    lifecycle_session.add(LifecycleRow(value=terminal_event))
+    lifecycle_session.flush()
+
+    getattr(lifecycle_session, terminal_event)()
+
+    assert lifecycle_session.info == {"unrelated": "preserved"}