txlog: address correctness, efficiency, and design review findings

native/src/txlog/arrow_ffi.rs

@@ -18,6 +18,24 @@ use arrow::record_batch::RecordBatch;
 use super::actions::FileEntry;
 use super::error::{TxLogError, Result};
 
+/// Convert a time-range stat value to epoch milliseconds for the Int64 Arrow column.
+///
+/// `AddAction::time_range_{start,end}` is stored as a String that may be either the
+/// legacy integer form (already epoch millis) or the Scala ISO-8601 form. Parsing
+/// with `i64::from_str` alone drops ISO-8601 values to null, losing time-range
+/// pruning data for Scala-written logs (TXLOG_MODULE_REVIEW D6).
+///
+/// The datetime-string formats (RFC 3339, bare datetime, date-only) are parsed by
+/// the shared `partition_pruning::parse_datetime_string_to_micros` so this and the
+/// stats-comparison path accept exactly the same set of formats.
+fn time_range_to_epoch_millis(s: &str) -> Option<i64> {
+    if let Ok(v) = s.parse::<i64>() {
+        return Some(v); // legacy epoch millis
+    }
+    // Shared parser returns micros; scale to the Int64 column's millisecond unit.
+    super::partition_pruning::parse_datetime_string_to_micros(s).map(|micros| micros / 1_000)
+}
+
 /// Build the Arrow schema dynamically based on partition columns and options.
 ///
 /// Base schema has 20 fixed columns. Additional columns are appended:
@@ -129,8 +147,8 @@ pub fn file_entries_to_record_batch(
         delete_opstamp_builder.append_option(add.delete_opstamp);
         num_merge_ops_builder.append_option(add.num_merge_ops);
         uncomp_size_builder.append_option(add.uncompressed_size_bytes);
-        time_start_builder.append_option(add.time_range_start.as_ref().and_then(|s| s.parse::<i64>().ok()));
-        time_end_builder.append_option(add.time_range_end.as_ref().and_then(|s| s.parse::<i64>().ok()));
+        time_start_builder.append_option(add.time_range_start.as_deref().and_then(time_range_to_epoch_millis));
+        time_end_builder.append_option(add.time_range_end.as_deref().and_then(time_range_to_epoch_millis));
         comp_delta_ver_builder.append_option(add.companion_delta_version);
 
         match &add.doc_mapping_json {

native/src/txlog/cache.rs

@@ -109,10 +109,10 @@ impl CacheConfig {
             cfg.file_list_capacity = n;
         }
 
-        // Concurrency
-        if let Some(n) = map.get("max_concurrent_reads").and_then(|s| s.parse::<usize>().ok()) {
-            cfg.max_concurrent_reads = if n == 0 { 32 } else { n };
-        }
+        // Concurrency — single source of truth for parsing this key, shared with
+        // the direct callers in distributed.rs (prevents the default/semantics from
+        // drifting between the two parse sites; see TXLOG_MODULE_REVIEW E4).
+        cfg.max_concurrent_reads = super::distributed::extract_max_concurrent(map);
 
         // Kill-switch
         if map.get("cache.enabled").map(|s| s.as_str()) == Some("false") {
@@ -502,6 +502,23 @@ pub fn get_or_create_cache(table_path: &str, config: CacheConfig) -> Arc<TxLogCa
         return cache.clone();
     }
 
+    // Reclaim idle registry entries so a long-running JVM that touches many distinct
+    // tables over time doesn't accumulate TxLogCache instances forever (each holds up
+    // to `snapshot_capacity` snapshots of Vec<FileEntry>). Once past the soft cap, drop
+    // entries no longer referenced by any live caller — `strong_count == 1` means only
+    // the registry holds it, so evicting is safe (it is recreated on the next access).
+    //
+    // This is a best-effort reclaim of *idle* tables, NOT a hard cap: if more than
+    // MAX_REGISTRY_ENTRIES tables are being queried concurrently, every entry has
+    // strong_count > 1 and nothing is freed. Bounding that case would require evicting
+    // live entries (causing duplicate caches for the same table) or a real LRU; the
+    // idle-reclaim covers the realistic "many tables over a long lifetime" pattern.
+    // See TXLOG_MODULE_REVIEW E4.
+    const MAX_REGISTRY_ENTRIES: usize = 256;
+    if registry.len() >= MAX_REGISTRY_ENTRIES {
+        registry.retain(|_, cache| Arc::strong_count(cache) > 1);
+    }
+
     let cache = Arc::new(TxLogCache::new(config));
     registry.insert(key, cache.clone());
     cache