From d230ed8fb3afa89b487fd70ae3a4db03bbabcb8b Mon Sep 17 00:00:00 2001 From: John Kerl Date: Wed, 8 Jul 2015 22:10:27 -0400 Subject: [PATCH] join iterate --- c/Makefile | 3 +++ c/input/lrec_reader_stdio_dkvp.c | 4 +-- c/input/lrec_readers.h | 2 ++ c/mapping/mapper_join.c | 44 +++++++++++++++----------------- 4 files changed, 28 insertions(+), 25 deletions(-) diff --git a/c/Makefile b/c/Makefile index 48d110b9c..c8d9f91ca 100644 --- a/c/Makefile +++ b/c/Makefile @@ -128,6 +128,9 @@ ap: evl: $(CCDEBUG) -D__LREC_EVALUATORS_MAIN__ mapping/lrec_evaluators.c mapping/mlr_val.c containers/mlr_dsl_ast.c containers/sllv.c containers/slls.c lib/mlrutil.c lib/mtrand.c containers/lrec.c -lm +lrim: + $(CCDEBUG) -DLREC_READER_IN_MEMORY_MAIN input/lrec_reader_in_memory.c containers/sllv.c containers/lrec.c containers/slls.c lib/mlrutil.c lib/mlr_globals.c input/lrec_reader_stdio_dkvp.c input/file_reader_stdio.c output/lrec_writer_dkvp.c -lm + # ---------------------------------------------------------------- clean: @rm -vf mlr mlrd mlrg mlrp tester diff --git a/c/input/lrec_reader_stdio_dkvp.c b/c/input/lrec_reader_stdio_dkvp.c index 2badfa323..594a0f14d 100644 --- a/c/input/lrec_reader_stdio_dkvp.c +++ b/c/input/lrec_reader_stdio_dkvp.c @@ -25,7 +25,7 @@ static lrec_t* lrec_reader_stdio_dkvp_process(void* pvhandle, void* pvstate, con } // No-op for stateless readers such as this one. -static void lrec_reader_csv_sof(void* pvstate) { +static void lrec_reader_stdio_dkvp_sof(void* pvstate) { } // No-op for stateless readers such as this one. @@ -45,7 +45,7 @@ lrec_reader_t* lrec_reader_stdio_dkvp_alloc(char irs, char ifs, char ips, int al plrec_reader->popen_func = &file_reader_stdio_vopen; plrec_reader->pclose_func = &file_reader_stdio_vclose; plrec_reader->pprocess_func = &lrec_reader_stdio_dkvp_process; - plrec_reader->psof_func = &lrec_reader_csv_sof; + plrec_reader->psof_func = &lrec_reader_stdio_dkvp_sof; plrec_reader->pfree_func = &lrec_reader_stdio_dkvp_free; return plrec_reader; diff --git a/c/input/lrec_readers.h b/c/input/lrec_readers.h index 8c4a4b152..7c3110204 100644 --- a/c/input/lrec_readers.h +++ b/c/input/lrec_readers.h @@ -19,6 +19,8 @@ lrec_reader_t* lrec_reader_mmap_dkvp_alloc(char irs, char ifs, char ips, int all lrec_reader_t* lrec_reader_mmap_nidx_alloc(char irs, char ifs, int allow_repeat_ifs); lrec_reader_t* lrec_reader_mmap_xtab_alloc(char irs, char ips, int allow_repeat_ips); +lrec_reader_t* lrec_reader_in_memory_alloc(sllv_t* precords); + // ---------------------------------------------------------------- // These entry points are made public for unit test lrec_t* lrec_parse_stdio_nidx(char* line, char ifs, int allow_repeat_ifs); diff --git a/c/mapping/mapper_join.c b/c/mapping/mapper_join.c index 8b39210f0..5f519069d 100644 --- a/c/mapping/mapper_join.c +++ b/c/mapping/mapper_join.c @@ -149,17 +149,6 @@ static int join_bucket_keeper_get_state(join_bucket_keeper_t* pkeeper) { // xxx put bucket & bucket-keeper into separate files with separate UTs static void join_bucket_keeper_initial_fill(join_bucket_keeper_t* pkeeper, slls_t* pleft_field_names) { - //pkeeper->pbucket; - //pkeeper->prec_peek; - //pkeeper->leof; - - //pkeeper->plrec_reader; - //pkeeper->pvhandle; - //pkeeper->pctx; - - //pbucket->pleft_field_values; - //pbucket->precords; - pkeeper->prec_peek = pkeeper->plrec_reader->pprocess_func(pkeeper->pvhandle, pkeeper->plrec_reader->pvstate, pkeeper->pctx); if (pkeeper->prec_peek == NULL) { @@ -188,6 +177,7 @@ static void join_bucket_keeper_initial_fill(join_bucket_keeper_t* pkeeper, slls_ sllv_add(pkeeper->pbucket->precords, pkeeper->prec_peek); pkeeper->prec_peek = NULL; } +} // if (pkeeper->leof) { // int cmp = slls_compare_lexically(pkeeper->pbucket->pjoin_values, pright_field_values); @@ -230,8 +220,6 @@ static void join_bucket_keeper_initial_fill(join_bucket_keeper_t* pkeeper, slls_ // sllv_add(pfoo, pleft_rec); // *ppbucket_paired = pfoo; -} - // xxx cmt re who frees static void join_bucket_keeper_emit(join_bucket_keeper_t* pkeeper, slls_t* pleft_field_names, slls_t* pright_field_values, @@ -247,22 +235,32 @@ static void join_bucket_keeper_emit(join_bucket_keeper_t* pkeeper, pkeeper->state = join_bucket_keeper_get_state(pkeeper); } +//typedef struct _join_bucket_t { +// slls_t* pleft_field_values; +// sllv_t* precords; +// int was_paired; +//} join_bucket_t; +// +//typedef struct _join_bucket_keeper_t { +// lrec_reader_t* plrec_reader; +// void* pvhandle; +// context_t* pctx; +// +// int state; +// join_bucket_t* pbucket; +// lrec_t* prec_peek; +// int leof; +//} join_bucket_keeper_t; + switch (pkeeper->state) { case LEFT_STATE_1_FULL: case LEFT_STATE_2_LAST_BUCKET: // Intentional fall-through cmp = slls_compare_lexically(pkeeper->pbucket->pleft_field_values, pright_field_values); if (cmp < 0) { - // Lunp <- bucket - // paired = null - // next state is 2 / 3 respectively + *ppbucket_left_unpaired = pkeeper->pbucket->precords; + // xxx advance left ... } else if (cmp == 0) { - // Lunp = null - // paired = bucket - // next state is 1 / 2 respectively - } else { - // Lunp = null - // paired = null - // next state is 1 / 2 respectively + *ppbucket_paired = pkeeper->pbucket->precords; } break;