Migrate common-storages-random.
@@ -26,7 +26,7 @@ use common_catalog::table::Table;
use common_catalog::table_context::TableContext;
use common_exception::Result;
use common_expression::Chunk;
-use common_expression::DataSchemaRef;
+use common_expression::ChunkEntry;
use common_expression::TableSchemaRef;
use common_meta_app::schema::TableInfo;
use common_pipeline_core::processors::port::OutputPort;
@@ -205,20 +205,27 @@ impl SyncSource for RandomSource {
return Ok(None);
}
- todo!("expression")
- // let columns = self
- // .schema
- // .fields()
- // .iter()
- // .map(|f| f.data_type().create_random_column(self.rows))
- // .collect();
- //
- // let rows = self.rows;
- //
- // // The partition garantees the number of rows is less than or equal to `max_block_size`.
- // // And we generate all the `self.rows` at once.
- // self.rows = 0;
- //
- // Ok(Some(Chunk::new(columns, rows)))
+ let columns = self
+ .schema
+ .fields()
+ .iter()
+ .enumerate()
+ .map(|(id, f)| {
+ let (col, ty) = f.data_type().create_random_column(self.rows);
+ ChunkEntry {
+ id,
+ data_type: ty,
+ value: col,
+ }
+ })
+ .collect();
+
+ let rows = self.rows;
+
+ // The partition garantees the number of rows is less than or equal to `max_block_size`.
+ // And we generate all the `self.rows` at once.
+ self.rows = 0;
+
+ Ok(Some(Chunk::new(columns, rows)))
}
}
