Jiadong latest recommendation gui

core/amber/requirements.txt

@@ -50,3 +50,5 @@ tenacity==8.5.0
 SQLAlchemy==2.0.37
 pg8000==1.31.2
 pympler==1.1
+tensorflow==2.19.0
+dataprofiler==0.13.3

core/amber/src/main/protobuf/edu/uci/ics/amber/engine/architecture/rpc/controlcommands.proto

@@ -57,6 +57,7 @@ message ControlRequest {
     EmptyRequest emptyRequest = 56;
     PrepareCheckpointRequest prepareCheckpointRequest = 57;
     QueryStatisticsRequest queryStatisticsRequest = 58;
+    QueryTableProfileRequest queryTableProfileRequest = 59;
 
     // request for testing
     Ping ping = 100;
@@ -271,4 +272,8 @@ message PrepareCheckpointRequest{
 
 message QueryStatisticsRequest{
   repeated core.ActorVirtualIdentity filterByWorkers = 1;
+}
+
+message QueryTableProfileRequest{
+
 }

core/amber/src/main/protobuf/edu/uci/ics/amber/engine/architecture/rpc/controlreturns.proto

@@ -19,6 +19,7 @@ syntax = "proto3";
 package edu.uci.ics.amber.engine.architecture.rpc;
 
 import "edu/uci/ics/amber/engine/architecture/worker/statistics.proto";
+import "edu/uci/ics/amber/engine/architecture/worker/tableprofile.proto";
 import "scalapb/scalapb.proto";
 
 option (scalapb.options) = {
@@ -43,6 +44,7 @@ message ControlReturn {
     WorkerStateResponse workerStateResponse = 50;
     WorkerMetricsResponse workerMetricsResponse = 51;
     FinalizeCheckpointResponse finalizeCheckpointResponse = 52;
+    TableProfileResponse tableProfileResponse = 53;
 
     // common responses
     ControlError controlError = 101;
@@ -137,4 +139,8 @@ message WorkerStateResponse {
 
 message WorkerMetricsResponse {
   worker.WorkerMetrics metrics = 1  [(scalapb.field).no_box = true];
+}
+
+message TableProfileResponse {
+  worker.TableProfile table_profiles = 1 [(scalapb.field).no_box = true];
 }

core/amber/src/main/protobuf/edu/uci/ics/amber/engine/architecture/rpc/workerservice.proto

@@ -40,6 +40,7 @@ service WorkerService {
   rpc PauseWorker(EmptyRequest) returns (WorkerStateResponse);
   rpc PrepareCheckpoint(PrepareCheckpointRequest) returns (EmptyReturn);
   rpc QueryStatistics(EmptyRequest) returns (WorkerMetricsResponse);
+  rpc QueryTableProfile(EmptyRequest) returns (TableProfileResponse);
   rpc ResumeWorker(EmptyRequest) returns (WorkerStateResponse);
   rpc RetrieveState(EmptyRequest) returns (EmptyReturn);
   rpc RetryCurrentTuple(EmptyRequest) returns (EmptyReturn);

core/amber/src/main/protobuf/edu/uci/ics/amber/engine/architecture/worker/tableprofile.proto

@@ -0,0 +1,142 @@
+// 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.
+
+syntax = "proto3";
+
+package edu.uci.ics.amber.engine.architecture.worker;
+
+import "scalapb/scalapb.proto";
+
+option (scalapb.options) = {
+  scope: FILE,
+  preserve_unknown_fields: false,
+  no_default_values_in_constructor: true
+};
+
+/* ------------------------------------------------------------------ */
+/*  GENERIC MATRIX (row-major flat array)                             */
+/* ------------------------------------------------------------------ */
+
+message NumericMatrix {
+  repeated double values = 1;  // length = rows * cols (row-major)
+  uint32 rows           = 2;
+  uint32 cols           = 3;
+}
+
+/* ------------------------------------------------------------------ */
+/*  GLOBAL PROFILE (dataset-wide)                                     */
+/* ------------------------------------------------------------------ */
+
+message GlobalProfile {
+
+  // ---- basic counts ----
+  uint64 samples_used        = 1;  // how many rows were sampled
+  uint64 column_count        = 2;
+  uint64 row_count           = 3;
+
+  // ---- row null / uniqueness ----
+  double row_has_null_ratio  = 4;
+  double row_is_null_ratio   = 5;
+  double unique_row_ratio    = 6;
+  uint64 duplicate_row_count = 7;
+
+  // ---- metadata ----
+  string file_type = 8;   // “csv”, “parquet”, …
+  string encoding  = 9;   // e.g. “utf-8”
+
+  // ---- pairwise stats ----
+  NumericMatrix correlation_matrix = 10;
+  NumericMatrix chi2_matrix        = 11;
+
+  // ---- schema map: column-name -> indices (to mirror DataProfiler) ---
+  map<string, ColumnIndexList> profile_schema = 12;
+
+  // ---- timing ----
+  message Times {
+    double row_stats_ms = 1;
+  }
+  Times times = 13;
+}
+
+// helper for profile_schema
+message ColumnIndexList {
+  repeated uint32 indices = 1;
+}
+
+message ColumnStatistics {
+
+  // ---- numeric summaries (nullable when not applicable) ----
+  double min                   = 1;
+  double max                   = 2;
+  double median                = 3;
+  double mean                  = 4;
+  double variance              = 5;
+  double stddev                = 6;
+  double skewness              = 7;
+  double kurtosis              = 8;
+  double sum                   = 9;
+
+  // ---- distribution ----
+  repeated double quantiles    = 10; // e.g. [q0, q0.5, q1]
+  uint64 num_zeros             = 11;
+  uint64 num_negatives         = 12;
+
+  // ---- uniqueness / cardinality ----
+  uint64 unique_count          = 13;
+  double unique_ratio          = 14;
+
+  // ---- categorical helpers ----
+  bool categorical             = 15;
+  map<string, uint64> categorical_count = 16;
+
+  // ---- nulls ----
+  uint64 null_count            = 17;
+  repeated string null_types   = 18;
+
+  // ---- data-type representation share (DataProfiler style) ----
+  map<string, double> data_type_representation = 19;
+}
+
+/* ------------------------------------------------------------------ */
+/*  FULL COLUMN PROFILE                                               */
+/* ------------------------------------------------------------------ */
+
+message ColumnProfile {
+
+  // identity
+  string column_name = 1;
+  string data_type   = 2;     // “string”, “int”, “float”, …
+  string data_label  = 3;
+  // quick hints
+  bool categorical   = 4;
+  string order       = 5;     // “random”, “ascending”, “constant value”
+
+  // examples
+  repeated string samples = 6;  // a few raw sample strings
+
+  // heavy stats
+  ColumnStatistics statistics = 7;
+}
+
+/* ------------------------------------------------------------------ */
+/*  TOP-LEVEL CONTAINER                                              */
+/* ------------------------------------------------------------------ */
+
+message TableProfile {
+  GlobalProfile global_profile            = 1;
+  repeated ColumnProfile column_profiles  = 2;
+}

core/amber/src/main/protobuf/edu/uci/ics/amber/engine/common/executionruntimestate.proto

@@ -22,6 +22,7 @@ package edu.uci.ics.amber.engine.common;
 import "edu/uci/ics/amber/engine/architecture/rpc/controlcommands.proto";
 import "edu/uci/ics/amber/engine/architecture/rpc/controlreturns.proto";
 import "edu/uci/ics/amber/engine/architecture/worker/statistics.proto";
+import "edu/uci/ics/amber/engine/architecture/worker/tableprofile.proto";
 import "edu/uci/ics/amber/core/virtualidentity.proto";
 import "edu/uci/ics/amber/core/workflowruntimestate.proto";
 import "scalapb/scalapb.proto";
@@ -88,7 +89,8 @@ message ExecutionStatsStore {
   int64 startTimeStamp = 1;
   int64 endTimeStamp = 2;
   map<string, OperatorMetrics> operator_info = 3;
-  repeated OperatorWorkerMapping operator_worker_mapping = 4;
+  map<string, architecture.worker.TableProfile> operator_table_profile = 4;
+  repeated OperatorWorkerMapping operator_worker_mapping = 5;
 }
 
 

core/amber/src/main/python/core/architecture/handlers/control/query_table_profile_handler.py

@@ -0,0 +1,22 @@
+#  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.
+from core.architecture.handlers.control.control_handler_base import ControlHandler
+from proto.edu.uci.ics.amber.engine.architecture.rpc import (EmptyRequest, TableProfileResponse)
+
+
+class QueryTableProfileHandler(ControlHandler):
+    async def query_table_profile(self, req: EmptyRequest) -> TableProfileResponse:
+        return TableProfileResponse(self.context.table_profile_manager.get_table_profile())

core/amber/src/main/python/core/architecture/managers/context.py

@@ -23,6 +23,7 @@
 from .debug_manager import DebugManager
 from .exception_manager import ExceptionManager
 from .state_processing_manager import StateProcessingManager
+from .table_profile_manager import TableProfileManager
 from .tuple_processing_manager import TupleProcessingManager
 from .executor_manager import ExecutorManager
 from .pause_manager import PauseManager
@@ -62,6 +63,7 @@ def __init__(self, worker_id, input_queue):
         )
 
         self.statistics_manager = StatisticsManager()
+        self.table_profile_manager = TableProfileManager()
         self.pause_manager = PauseManager(
             self.input_queue, state_manager=self.state_manager
         )