fix bug rec_len diff parsed_len in pm_ppl_recv_parse_log_rec(). How I fixed: save the log_ptr offset in page_copy_rec_list_end_to_created_page() and retrieve it.

Author: trdtnguyen

storage/innobase/btr/btr0btr.cc

@@ -2147,6 +2147,9 @@ btr_insert_on_non_leaf_level_func(
 		tuple, &rec, &dummy_big_rec, 0, NULL, mtr);
 
 	if (err == DB_FAIL) {
+//#if defined (UNIV_PMEMOBJ_PART_PL)
+//		printf("PMEM_DEBUG split chain case mtr %zu\n", mtr);
+//#endif
 		err = btr_cur_pessimistic_insert(flags
 						 | BTR_NO_LOCKING_FLAG
 						 | BTR_KEEP_SYS_FLAG

storage/innobase/btr/btr0cur.cc

@@ -4918,7 +4918,8 @@ btr_cur_del_mark_set_sec_rec_log(
 	byte*	log_ptr;
 	ut_ad(val <= 1);
 
-	log_ptr = mlog_open(mtr, 11 + 1 + 2);
+	//log_ptr = mlog_open(mtr, 11 + 1 + 2);
+	log_ptr = mlog_open(mtr, MLOG_HEADER_SIZE + 1 + 2);
 
 	if (!log_ptr) {
 		/* Logging in mtr is switched off during crash recovery:

storage/innobase/fil/fil0fil.cc

@@ -1428,7 +1428,9 @@ fil_space_create(
 	}
 
 	mutex_exit(&fil_system->mutex);
-
+//tdnguyen test
+	printf("=====>>|| fil_space_create() space id %zu name %s\n", id, name);
+//end tdnguyen test
 	return(space);
 }
 
@@ -2116,7 +2118,8 @@ fil_op_write_log(
 	ut_ad(strchr(path, OS_PATH_SEPARATOR) != NULL);
 	ut_ad(strcmp(&path[strlen(path) - strlen(DOT_IBD)], DOT_IBD) == 0);
 
-	log_ptr = mlog_open(mtr, 11 + 4 + 2 + 1);
+	//log_ptr = mlog_open(mtr, 11 + 4 + 2 + 1);
+	log_ptr = mlog_open(mtr, MLOG_HEADER_SIZE + 4 + 2 + 1);
 
 	if (log_ptr == NULL) {
 		/* Logging in mtr is switched off during crash recovery:
@@ -8727,22 +8730,6 @@ pm_log_fil_node_create(
 		assert(0);
 		return NULL;
 	}
-
-//	//open file, simple version of fil_node_open_file() in fil0fil.cc
-//	node->handle = os_file_create(
-//			innodb_log_file_key, node->name, OS_FILE_OPEN,
-//			OS_FILE_AIO, OS_LOG_FILE, false, &success);	   
-//
-//	if (!success){
-//		printf("PMEM_ERROR in pm_create_or_open_part_log_files(), cannot open fil_node %s \n ", node->name);
-//		assert(0);
-//	}
-//
-//	node->is_open = true;
-//
-//	fil_system->n_open++;
-//	fil_n_file_opened++;
-
 	return node;
 }
 /*
@@ -8757,7 +8744,6 @@ pm_log_fil_io(
 {
 
 	PMEM_LOG_GROUP*		group;
-	PMEM_PAGE_LOG_HASHED_LINE* pline;
 
 	byte*		pdata;
 	byte*		log_src;
@@ -8802,20 +8788,24 @@ pm_log_fil_io(
 	assert(plogbuf->hashed_id >= 0);	
 
 	group = ppl->log_groups[plogbuf->hashed_id];
-	pline = D_RW(D_RW(ppl->buckets)[plogbuf->hashed_id]);
-
-	next_offset = pline->write_diskaddr;
 
-	if (next_offset + len > group->file_size){
-		printf("PMEM_INFO log file of line %zu is full\n", plogbuf->hashed_id);
-		assert(0);
-		//TODO: handle the case that the log group is full
-		//write_len = group->file_size - next_offset;
-	}
-	else {
-		write_len = len;
-	}
+	/*if the previous AIO is not finish while this AIO happen, write_diskaddr < diskaddr. 
+	 * Otherwise, write_diskaddr == diskaddr. 
+	 * So we write from diskaddr to avoid overwirte on previous written
+	 * Since we advanced diskaddr size-bytes when handling full pline, we write from the previous size bytes from diskaddr
+	 * */
 
+	//next_offset = pline->write_diskaddr
+	next_offset = plogbuf->diskaddr;
+	
+	//tdnguyen test, fixing bug pline 593 has first write block zero
+	
+	//what if we only write the first block	
+	if (plogbuf->hashed_id == 593 && next_offset > plogbuf->size ) {
+		printf("===> TEST only write first block ...\n\n");
+		return;		
+	}
+
 	//(2) Write, simulate log_group_write_buf()
 	assert((end_offset % univ_page_size.physical()) == 0);	
 	//ut_ad(len % OS_FILE_LOG_BLOCK_SIZE == 0);
@@ -8833,6 +8823,25 @@ pm_log_fil_io(
 		}
 		node->is_open = true;
 	}
+
+	if (next_offset + len > group->file_size){
+		printf("PMEM_INFO log file of line %zu is full, extend it to double size\n", plogbuf->hashed_id);
+		//assert(0);
+		bool ret = os_file_truncate(
+			   	node->name,
+				node->handle,
+			   	group->file_size * 2);
+		if (!ret){
+			printf("PMEM_ERROR cannot extend log file of line %zu \n", plogbuf->hashed_id);	
+			assert(0);
+		}
+
+		group->file_size = group->file_size * 2;
+
+	}
+	else {
+		write_len = len;
+	}
 
 	err = os_aio(
 		req_type,

storage/innobase/handler/handler0alter.cc

@@ -8449,6 +8449,10 @@ ha_innobase::commit_inplace_alter_table(
 			ut_ad(trx_is_rseg_updated(trx));
 
 			if (mtr.get_log()->size() > 0) {
+#if defined (UNIV_PMEMOBJ_PART_PL)
+				//TODO: handle this
+				assert(0);
+#endif
 				ut_ad(*mtr.get_log()->front()->begin()
 				      == MLOG_FILE_RENAME2);
 

storage/innobase/include/dyn0buf.h

@@ -372,6 +372,25 @@ class dyn_buf_t {
 		return(true);
 	}
 
+#if defined (UNIV_PMEMOBJ_PART_PL)
+	/**
+	@return the next block */
+	block_t* next(block_t* block)
+		MY_ATTRIBUTE((warn_unused_result))
+	{
+		ut_ad(UT_LIST_GET_LEN(m_list) > 0);
+		if (block == NULL){
+			return NULL;
+		}
+
+		return(UT_LIST_GET_NEXT(m_node, block));
+	}
+
+	block_t* get_back(){
+		return (UT_LIST_GET_LAST(m_list));
+	}
+#endif
+
 	/**
 	@return the first block */
 	block_t* front()

storage/innobase/include/mtr0log.h

@@ -84,8 +84,18 @@ mlog_write_initial_log_record(
 				modification is made */
 	mlog_id_t	type,	/*!< in: log item type: MLOG_1BYTE, ... */
 	mtr_t*		mtr);	/*!< in: mini-transaction handle */
+
 /********************************************************//**
 Catenates 1 - 4 bytes to the mtr log. The value is not compressed. */
+#if defined (UNIV_PMEMOBJ_PART_PL)
+UNIV_INLINE
+void
+mlog_catenate_ulint(
+/*================*/
+	mtr_buf_t*	dyn_buf,	/*!< in/out: buffer to write */
+	ulint		val,		/*!< in: value to write */
+	mlog_id_t	type);		/*!< in: type of value to write */
+#else
 UNIV_INLINE
 void
 mlog_catenate_ulint(
@@ -102,6 +112,7 @@ mlog_catenate_ulint(
 	mtr_t*		mtr,	/*!< in: mtr */
 	ulint		val,	/*!< in: value to write */
 	mlog_id_t	type);	/*!< in: MLOG_1BYTE, MLOG_2BYTES, MLOG_4BYTES */
+#endif
 /********************************************************//**
 Catenates n bytes to the mtr log. */
 void