fix(stream): let Row Merge executor handle extra row count column in input

fix(stream): let Row Merge executor handle extra row count column in input #5432